/*

 * CDDL HEADER START

 *

 * The contents of this file are subject to the terms of the

 * Common Development and Distribution License, Version 1.0 only

 * (the "License").  You may not use this file except in compliance

 * with the License.

 *

 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

 * or http://www.opensolaris.org/os/licensing.

 * See the License for the specific language governing permissions

 * and limitations under the License.

 *

 * When distributing Covered Code, include this CDDL HEADER in each

 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.

 * If applicable, add the following below this CDDL HEADER, with the

 * fields enclosed by brackets "[]" replaced with your own identifying

 * information: Portions Copyright [yyyy] [name of copyright owner]

 *

 * CDDL HEADER END

 */

/*

 * Copyright 2005 Sun Microsystems, Inc.  All rights reserved.

 * Use is subject to license terms.

 */

#pragma ident	"%Z%%M%	%I%	%E% SMI"

#include <sys/types.h>

#include <sys/stream.h>

#include <sys/strsun.h>

#include <sys/strsubr.h>

#include <sys/debug.h>

#include <sys/cmn_err.h>

#include <sys/tihdr.h>

#include <inet/common.h>

#include <inet/ip.h>

#include <inet/ip_impl.h>

#include <inet/tcp.h>

#include <inet/tcp_impl.h>

#include <inet/ipsec_impl.h>

#include <inet/ipclassifier.h>

#include <inet/ipp_common.h>

/*

 * This file implements TCP fusion - a protocol-less data path for TCP

 * loopback connections.  The fusion of two local TCP endpoints occurs

 * at connection establishment time.  Various conditions (see details

 * in tcp_fuse()) need to be met for fusion to be successful.  If it

 * fails, we fall back to the regular TCP data path; if it succeeds,

 * both endpoints proceed to use tcp_fuse_output() as the transmit path.

 * tcp_fuse_output() enqueues application data directly onto the peer's

 * receive queue; no protocol processing is involved.  After enqueueing

 * the data, the sender can either push (putnext) data up the receiver's

 * read queue; or the sender can simply return and let the receiver

 * retrieve the enqueued data via the synchronous streams entry point

 * tcp_fuse_rrw().  The latter path is taken if synchronous streams is

 * enabled (the default).  It is disabled if sockfs no longer resides

 * directly on top of tcp module due to a module insertion or removal.

 * It also needs to be temporarily disabled when sending urgent data

 * because the tcp_fuse_rrw() path bypasses the M_PROTO processing done

 * by strsock_proto() hook.

 *

 * Sychronization is handled by squeue and the mutex tcp_fuse_lock.

 * One of the requirements for fusion to succeed is that both endpoints

 * need to be using the same squeue.  This ensures that neither side

 * can disappear while the other side is still sending data.  By itself,

 * squeue is not sufficient for guaranteeing safety when synchronous

 * streams is enabled.  The reason is that tcp_fuse_rrw() doesn't enter

 * the squeue and its access to tcp_rcv_list and other fusion-related

 * fields needs to be sychronized with the sender.  tcp_fuse_lock is

 * used for this purpose.  When there is urgent data, the sender needs

 * to push the data up the receiver's streams read queue.  In order to

 * avoid holding the tcp_fuse_lock across putnext(), the sender sets

 * the peer tcp's tcp_fuse_syncstr_stopped bit and releases tcp_fuse_lock

 * (see macro TCP_FUSE_SYNCSTR_STOP()).  If tcp_fuse_rrw() enters after

 * this point, it will see that synchronous streams is temporarily

 * stopped and it will immediately return EBUSY without accessing the

 * tcp_rcv_list or other fields protected by the tcp_fuse_lock.  This

 * will result in strget() calling getq_noenab() to dequeue data from

 * the stream head instead.  After the sender has finished pushing up

 * all urgent data, it will clear the tcp_fuse_syncstr_stopped bit using

 * TCP_FUSE_SYNCSTR_RESUME and the receiver may then resume using

 * tcp_fuse_rrw() to retrieve data from tcp_rcv_list.

 *

 * The following note applies only to the synchronous streams mode.

 *

 * Flow control is done by checking the size of receive buffer and

 * the number of data blocks, both set to different limits.  This is

 * different than regular streams flow control where cumulative size

 * check dominates block count check -- streams queue high water mark

 * typically represents bytes.  Each enqueue triggers notifications

 * to the receiving process; a build up of data blocks indicates a

 * slow receiver and the sender should be blocked or informed at the

 * earliest moment instead of further wasting system resources.  In

 * effect, this is equivalent to limiting the number of outstanding

 * segments in flight.

 */

/*

 * Macros that determine whether or not IP processing is needed for TCP.

 */

#define	TCP_IPOPT_POLICY_V4(tcp)					\

	((tcp)->tcp_ipversion == IPV4_VERSION &&			\

	((tcp)->tcp_ip_hdr_len != IP_SIMPLE_HDR_LENGTH ||		\

	CONN_OUTBOUND_POLICY_PRESENT((tcp)->tcp_connp) ||		\

	CONN_INBOUND_POLICY_PRESENT((tcp)->tcp_connp)))

#define	TCP_IPOPT_POLICY_V6(tcp)					\

	((tcp)->tcp_ipversion == IPV6_VERSION &&			\

	((tcp)->tcp_ip_hdr_len != IPV6_HDR_LEN ||			\

	CONN_OUTBOUND_POLICY_PRESENT_V6((tcp)->tcp_connp) ||		\

	CONN_INBOUND_POLICY_PRESENT_V6((tcp)->tcp_connp)))

#define	TCP_LOOPBACK_IP(tcp)						\

	(TCP_IPOPT_POLICY_V4(tcp) || TCP_IPOPT_POLICY_V6(tcp) ||	\

	!CONN_IS_MD_FASTPATH((tcp)->tcp_connp))

/*

 * Setting this to false means we disable fusion altogether and

 * loopback connections would go through the protocol paths.

 */

boolean_t do_tcp_fusion = B_TRUE;

/*

 * Enabling this flag allows sockfs to retrieve data directly

 * from a fused tcp endpoint using synchronous streams interface.

 */

boolean_t do_tcp_direct_sockfs = B_TRUE;

/*

 * This is the minimum amount of outstanding writes allowed on

 * a synchronous streams-enabled receiving endpoint before the

 * sender gets flow-controlled.  Setting this value to 0 means

 * that the data block limit is equivalent to the byte count

 * limit, which essentially disables the check.

 */

#define	TCP_FUSION_RCV_UNREAD_MIN	8

uint_t tcp_fusion_rcv_unread_min = TCP_FUSION_RCV_UNREAD_MIN;

static void	tcp_fuse_syncstr_enable(tcp_t *);

static void	tcp_fuse_syncstr_disable(tcp_t *);

static void	strrput_sig(queue_t *, boolean_t);

/*

 * This routine gets called by the eager tcp upon changing state from

 * SYN_RCVD to ESTABLISHED.  It fuses a direct path between itself

 * and the active connect tcp such that the regular tcp processings

 * may be bypassed under allowable circumstances.  Because the fusion

 * requires both endpoints to be in the same squeue, it does not work

 * for simultaneous active connects because there is no easy way to

 * switch from one squeue to another once the connection is created.

 * This is different from the eager tcp case where we assign it the

 * same squeue as the one given to the active connect tcp during open.

 */

void

tcp_fuse(tcp_t *tcp, uchar_t *iphdr, tcph_t *tcph)

{

	conn_t *peer_connp, *connp = tcp->tcp_connp;

	tcp_t *peer_tcp;

	ASSERT(!tcp->tcp_fused);

	ASSERT(tcp->tcp_loopback);

	ASSERT(tcp->tcp_loopback_peer == NULL);

	/*

	 * We need to inherit q_hiwat of the listener tcp, but we can't

	 * really use tcp_listener since we get here after sending up

	 * T_CONN_IND and tcp_wput_accept() may be called independently,

	 * at which point tcp_listener is cleared; this is why we use

	 * tcp_saved_listener.  The listener itself is guaranteed to be

	 * around until tcp_accept_finish() is called on this eager --

	 * this won't happen until we're done since we're inside the

	 * eager's perimeter now.

	 */

	ASSERT(tcp->tcp_saved_listener != NULL);

	/*

	 * Lookup peer endpoint; search for the remote endpoint having

	 * the reversed address-port quadruplet in ESTABLISHED state,

	 * which is guaranteed to be unique in the system.  Zone check

	 * is applied accordingly for loopback address, but not for

	 * local address since we want fusion to happen across Zones.

	 */

	if (tcp->tcp_ipversion == IPV4_VERSION) {

		peer_connp = ipcl_conn_tcp_lookup_reversed_ipv4(connp,

		    (ipha_t *)iphdr, tcph);

	} else {

		peer_connp = ipcl_conn_tcp_lookup_reversed_ipv6(connp,

		    (ip6_t *)iphdr, tcph);

	}

	/*

	 * We can only proceed if peer exists, resides in the same squeue

	 * as our conn and is not raw-socket.  The squeue assignment of

	 * this eager tcp was done earlier at the time of SYN processing

	 * in ip_fanout_tcp{_v6}.  Note that similar squeues by itself

	 * doesn't guarantee a safe condition to fuse, hence we perform

	 * additional tests below.

	 */

	ASSERT(peer_connp == NULL || peer_connp != connp);

	if (peer_connp == NULL || peer_connp->conn_sqp != connp->conn_sqp ||

	    !IPCL_IS_TCP(peer_connp)) {

		if (peer_connp != NULL) {

			TCP_STAT(tcp_fusion_unqualified);

			CONN_DEC_REF(peer_connp);

		}

		return;

	}

	peer_tcp = peer_connp->conn_tcp;	/* active connect tcp */

	ASSERT(peer_tcp != NULL && peer_tcp != tcp && !peer_tcp->tcp_fused);

	ASSERT(peer_tcp->tcp_loopback && peer_tcp->tcp_loopback_peer == NULL);

	ASSERT(peer_connp->conn_sqp == connp->conn_sqp);

	/*

	 * Fuse the endpoints; we perform further checks against both

	 * tcp endpoints to ensure that a fusion is allowed to happen.

	 * In particular we bail out for non-simple TCP/IP or if IPsec/

	 * IPQoS policy exists.

	 */

	if (!tcp->tcp_unfusable && !peer_tcp->tcp_unfusable &&

	    !TCP_LOOPBACK_IP(tcp) && !TCP_LOOPBACK_IP(peer_tcp) &&

	    !IPP_ENABLED(IPP_LOCAL_OUT|IPP_LOCAL_IN)) {

		mblk_t *mp;

		struct stroptions *stropt;

		queue_t *peer_rq = peer_tcp->tcp_rq;

		ASSERT(!TCP_IS_DETACHED(peer_tcp) && peer_rq != NULL);

		ASSERT(tcp->tcp_fused_sigurg_mp == NULL);

		ASSERT(peer_tcp->tcp_fused_sigurg_mp == NULL);

		/*

		 * We need to drain data on both endpoints during unfuse.

		 * If we need to send up SIGURG at the time of draining,

		 * we want to be sure that an mblk is readily available.

		 * This is why we pre-allocate the M_PCSIG mblks for both

		 * endpoints which will only be used during/after unfuse.

		 */

		if ((mp = allocb(1, BPRI_HI)) == NULL)

			goto failed;

		tcp->tcp_fused_sigurg_mp = mp;

		if ((mp = allocb(1, BPRI_HI)) == NULL)

			goto failed;

		peer_tcp->tcp_fused_sigurg_mp = mp;

		/* Allocate M_SETOPTS mblk */

		if ((mp = allocb(sizeof (*stropt), BPRI_HI)) == NULL)

			goto failed;

		/* Fuse both endpoints */

		peer_tcp->tcp_loopback_peer = tcp;

		tcp->tcp_loopback_peer = peer_tcp;

		peer_tcp->tcp_fused = tcp->tcp_fused = B_TRUE;

		/*

		 * We never use regular tcp paths in fusion and should

		 * therefore clear tcp_unsent on both endpoints.  Having

		 * them set to non-zero values means asking for trouble

		 * especially after unfuse, where we may end up sending

		 * through regular tcp paths which expect xmit_list and

		 * friends to be correctly setup.

		 */

		peer_tcp->tcp_unsent = tcp->tcp_unsent = 0;

		tcp_timers_stop(tcp);

		tcp_timers_stop(peer_tcp);

		/*

		 * At this point we are a detached eager tcp and therefore

		 * don't have a queue assigned to us until accept happens.

		 * In the mean time the peer endpoint may immediately send

		 * us data as soon as fusion is finished, and we need to be

		 * able to flow control it in case it sends down huge amount

		 * of data while we're still detached.  To prevent that we

		 * inherit the listener's q_hiwat value; this is temporary

		 * since we'll repeat the process in tcp_accept_finish().

		 */

		(void) tcp_fuse_set_rcv_hiwat(tcp,

		    tcp->tcp_saved_listener->tcp_rq->q_hiwat);

		/*

		 * Set the stream head's write offset value to zero since we

		 * won't be needing any room for TCP/IP headers; tell it to

		 * not break up the writes (this would reduce the amount of

		 * work done by kmem); and configure our receive buffer.

		 * Note that we can only do this for the active connect tcp

		 * since our eager is still detached; it will be dealt with

		 * later in tcp_accept_finish().

		 */

		DB_TYPE(mp) = M_SETOPTS;

		mp->b_wptr += sizeof (*stropt);

		stropt = (struct stroptions *)mp->b_rptr;

		stropt->so_flags = SO_MAXBLK | SO_WROFF | SO_HIWAT;

		stropt->so_maxblk = tcp_maxpsz_set(peer_tcp, B_FALSE);

		stropt->so_wroff = 0;

		/*

		 * Record the stream head's high water mark for

		 * peer endpoint; this is used for flow-control

		 * purposes in tcp_fuse_output().

		 */

		stropt->so_hiwat = tcp_fuse_set_rcv_hiwat(peer_tcp,

		    peer_rq->q_hiwat);

		/* Send the options up */

		putnext(peer_rq, mp);

	} else {

		TCP_STAT(tcp_fusion_unqualified);

	}

	CONN_DEC_REF(peer_connp);

	return;

failed:

	if (tcp->tcp_fused_sigurg_mp != NULL) {

		freeb(tcp->tcp_fused_sigurg_mp);

		tcp->tcp_fused_sigurg_mp = NULL;

	}

	if (peer_tcp->tcp_fused_sigurg_mp != NULL) {

		freeb(peer_tcp->tcp_fused_sigurg_mp);

		peer_tcp->tcp_fused_sigurg_mp = NULL;

	}

	CONN_DEC_REF(peer_connp);

}

/*

 * Unfuse a previously-fused pair of tcp loopback endpoints.

 */

void

tcp_unfuse(tcp_t *tcp)

{

	tcp_t *peer_tcp = tcp->tcp_loopback_peer;

	ASSERT(tcp->tcp_fused && peer_tcp != NULL);

	ASSERT(peer_tcp->tcp_fused && peer_tcp->tcp_loopback_peer == tcp);

	ASSERT(tcp->tcp_connp->conn_sqp == peer_tcp->tcp_connp->conn_sqp);

	ASSERT(tcp->tcp_unsent == 0 && peer_tcp->tcp_unsent == 0);

	ASSERT(tcp->tcp_fused_sigurg_mp != NULL);

	ASSERT(peer_tcp->tcp_fused_sigurg_mp != NULL);

	/*

	 * We disable synchronous streams, drain any queued data and

	 * clear tcp_direct_sockfs.  The synchronous streams entry

	 * points will become no-ops after this point.

	 */

	tcp_fuse_disable_pair(tcp, B_TRUE);

	/*

	 * Update th_seq and th_ack in the header template

	 */

	U32_TO_ABE32(tcp->tcp_snxt, tcp->tcp_tcph->th_seq);

	U32_TO_ABE32(tcp->tcp_rnxt, tcp->tcp_tcph->th_ack);

	U32_TO_ABE32(peer_tcp->tcp_snxt, peer_tcp->tcp_tcph->th_seq);

	U32_TO_ABE32(peer_tcp->tcp_rnxt, peer_tcp->tcp_tcph->th_ack);

	/* Unfuse the endpoints */

	peer_tcp->tcp_fused = tcp->tcp_fused = B_FALSE;

	peer_tcp->tcp_loopback_peer = tcp->tcp_loopback_peer = NULL;

}

/*

 * Fusion output routine for urgent data.  This routine is called by

 * tcp_fuse_output() for handling non-M_DATA mblks.

 */

void

tcp_fuse_output_urg(tcp_t *tcp, mblk_t *mp)

{

	mblk_t *mp1;

	struct T_exdata_ind *tei;

	tcp_t *peer_tcp = tcp->tcp_loopback_peer;

	mblk_t *head, *prev_head = NULL;

	ASSERT(tcp->tcp_fused);

	ASSERT(peer_tcp != NULL && peer_tcp->tcp_loopback_peer == tcp);

	ASSERT(DB_TYPE(mp) == M_PROTO || DB_TYPE(mp) == M_PCPROTO);

	ASSERT(mp->b_cont != NULL && DB_TYPE(mp->b_cont) == M_DATA);

	ASSERT(MBLKL(mp) >= sizeof (*tei) && MBLKL(mp->b_cont) > 0);

	/*

	 * Urgent data arrives in the form of T_EXDATA_REQ from above.

	 * Each occurence denotes a new urgent pointer.  For each new

	 * urgent pointer we signal (SIGURG) the receiving app to indicate

	 * that it needs to go into urgent mode.  This is similar to the

	 * urgent data handling in the regular tcp.  We don't need to keep

	 * track of where the urgent pointer is, because each T_EXDATA_REQ

	 * "advances" the urgent pointer for us.

	 *

	 * The actual urgent data carried by T_EXDATA_REQ is then prepended

	 * by a T_EXDATA_IND before being enqueued behind any existing data

	 * destined for the receiving app.  There is only a single urgent

	 * pointer (out-of-band mark) for a given tcp.  If the new urgent

	 * data arrives before the receiving app reads some existing urgent

	 * data, the previous marker is lost.  This behavior is emulated

	 * accordingly below, by removing any existing T_EXDATA_IND messages

	 * and essentially converting old urgent data into non-urgent.

	 */

	ASSERT(tcp->tcp_valid_bits & TCP_URG_VALID);

	/* Let sender get out of urgent mode */

	tcp->tcp_valid_bits &= ~TCP_URG_VALID;

	/*

	 * This flag indicates that a signal needs to be sent up.

	 * This flag will only get cleared once SIGURG is delivered and

	 * is not affected by the tcp_fused flag -- delivery will still

	 * happen even after an endpoint is unfused, to handle the case

	 * where the sending endpoint immediately closes/unfuses after

	 * sending urgent data and the accept is not yet finished.

	 */

	peer_tcp->tcp_fused_sigurg = B_TRUE;

	/* Reuse T_EXDATA_REQ mblk for T_EXDATA_IND */

	DB_TYPE(mp) = M_PROTO;

	tei = (struct T_exdata_ind *)mp->b_rptr;

	tei->PRIM_type = T_EXDATA_IND;

	tei->MORE_flag = 0;

	mp->b_wptr = (uchar_t *)&tei[1];

	TCP_STAT(tcp_fusion_urg);

	BUMP_MIB(&tcp_mib, tcpOutUrg);

	head = peer_tcp->tcp_rcv_list;

	while (head != NULL) {

		/*

		 * Remove existing T_EXDATA_IND, keep the data which follows

		 * it and relink our list.  Note that we don't modify the

		 * tcp_rcv_last_tail since it never points to T_EXDATA_IND.

		 */

		if (DB_TYPE(head) != M_DATA) {

			mp1 = head;

			ASSERT(DB_TYPE(mp1->b_cont) == M_DATA);

			head = mp1->b_cont;

			mp1->b_cont = NULL;

			head->b_next = mp1->b_next;

			mp1->b_next = NULL;

			if (prev_head != NULL)

				prev_head->b_next = head;

			if (peer_tcp->tcp_rcv_list == mp1)

				peer_tcp->tcp_rcv_list = head;

			if (peer_tcp->tcp_rcv_last_head == mp1)

				peer_tcp->tcp_rcv_last_head = head;

			freeb(mp1);

		}

		prev_head = head;

		head = head->b_next;

	}

}

/*

 * Fusion output routine, called by tcp_output() and tcp_wput_proto().

 */

boolean_t

tcp_fuse_output(tcp_t *tcp, mblk_t *mp, uint32_t send_size)

{

	tcp_t *peer_tcp = tcp->tcp_loopback_peer;

	queue_t *peer_rq;

	uint_t max_unread;

	boolean_t flow_stopped;

	boolean_t urgent = (DB_TYPE(mp) != M_DATA);

	ASSERT(tcp->tcp_fused);

	ASSERT(peer_tcp != NULL && peer_tcp->tcp_loopback_peer == tcp);

	ASSERT(tcp->tcp_connp->conn_sqp == peer_tcp->tcp_connp->conn_sqp);

	ASSERT(DB_TYPE(mp) == M_DATA || DB_TYPE(mp) == M_PROTO ||

	    DB_TYPE(mp) == M_PCPROTO);

	peer_rq = peer_tcp->tcp_rq;

	max_unread = peer_tcp->tcp_fuse_rcv_unread_hiwater;

	/* If this connection requires IP, unfuse and use regular path */

	if (TCP_LOOPBACK_IP(tcp) || TCP_LOOPBACK_IP(peer_tcp) ||

	    IPP_ENABLED(IPP_LOCAL_OUT|IPP_LOCAL_IN)) {

		TCP_STAT(tcp_fusion_aborted);

		tcp_unfuse(tcp);

		return (B_FALSE);

	}

	if (send_size == 0) {

		freemsg(mp);

		return (B_TRUE);

	}

	/*

	 * Handle urgent data; we either send up SIGURG to the peer now

	 * or do it later when we drain, in case the peer is detached

	 * or if we're short of memory for M_PCSIG mblk.

	 */

	if (urgent) {

		/*

		 * We stop synchronous streams when we have urgent data

		 * queued to prevent tcp_fuse_rrw() from pulling it.  If

		 * for some reasons the urgent data can't be delivered

		 * below, synchronous streams will remain stopped until

		 * someone drains the tcp_rcv_list.

		 */

		TCP_FUSE_SYNCSTR_STOP(peer_tcp);

		tcp_fuse_output_urg(tcp, mp);

	}

	mutex_enter(&peer_tcp->tcp_fuse_lock);

	/*

	 * Wake up and signal the peer; it is okay to do this before

	 * enqueueing because we are holding the lock.  One of the

	 * advantages of synchronous streams is the ability for us to

	 * find out when the application performs a read on the socket,

	 * by way of tcp_fuse_rrw() entry point being called.  Every

	 * data that gets enqueued onto the receiver is treated as if

	 * it has arrived at the receiving endpoint, thus generating

	 * SIGPOLL/SIGIO for asynchronous socket just as in the strrput()

	 * case.  However, we only wake up the application when necessary,

	 * i.e. during the first enqueue.  When tcp_fuse_rrw() is called

	 * it will send everything upstream.

	 */

	if (peer_tcp->tcp_direct_sockfs && !urgent &&

	    !TCP_IS_DETACHED(peer_tcp)) {

		if (peer_tcp->tcp_rcv_list == NULL)

			STR_WAKEUP_SET(STREAM(peer_tcp->tcp_rq));

		/* Update poll events and send SIGPOLL/SIGIO if necessary */

		STR_SENDSIG(STREAM(peer_tcp->tcp_rq));

	}

	/*

	 * Enqueue data into the peer's receive list; we may or may not

	 * drain the contents depending on the conditions below.

	 */