// :vi sw=4 ts=4 noet:
/*
==================================================================================
	Copyright (c) 2019 Nokia
	Copyright (c) 2018-2019 AT&T Intellectual Property.

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

	   http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
==================================================================================
*/

/*
	Mnemonic:	sr_static.c
	Abstract:	These are static send/receive related functions.

				(broken out of rmr.c)
	Author:		E. Scott Daniels
	Date:		13 February 2019
*/

#ifndef _sr_static_c
#define _sr_static_c

/*
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <netdb.h>
#include <errno.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <unistd.h>
#include <stdint.h>
#include <time.h>
#include <arpa/inet.h>

#include <nanomsg/nn.h>
#include <nanomsg/tcp.h>
#include <nanomsg/pair.h>
#include <nanomsg/pipeline.h>
#include <nanomsg/pubsub.h>

#include "rmr.h"				// things the users see
#include "rmr_private.h"		// things that we need too
#include "rmr_symtab.h"

#include "ring_static.c"		// message ring support
#include "rt_generic_static.c"	// generic route table (not nng/nano specific)
#include "rtable_static.c"		// route table things	(nano specific)
#include "tools_static.c"
*/


/*
	Alloc a new nano zero copy buffer and put into msg. If msg is nil, then we will alloc
	a new message struct as well. Size is the size of the zc buffer to allocate (not
	including our header). If size is 0, then the buffer allocated is the size previously
	allocated (if msg is !nil) or the default size given at initialisation).


	The trlo parm is the trace length override which will be used if not 0. If 0, then the
	length in the context is used (default).
*/
static rmr_mbuf_t* alloc_zcmsg( uta_ctx_t* ctx, rmr_mbuf_t* msg, int size, int state, int trlo ) {
	int	mlen;
	uta_mhdr_t*	hdr;
	int tr_len;				// length to allocate for trace info

	tr_len = trlo > 0 ? trlo : ctx->trace_data_len;

	mlen = sizeof( uta_mhdr_t ) + tr_len + ctx->d1_len + ctx->d2_len;	// start with header and trace/data lengths
	mlen += (size > 0 ? size  : ctx->max_plen);							// add user requested size or size set during init

	if( msg == NULL ) {
		msg = (rmr_mbuf_t *) malloc( sizeof *msg );
		if( msg == NULL ) {
			fprintf( stderr, "[CRIT] rmr_alloc_zc: cannot get memory for message\n" );
			exit( 1 );
		}
	} else {
		mlen = msg->alloc_len;							// msg given, allocate the same size as before
	}

	memset( msg, 0, sizeof( *msg ) );

	if( (msg->header = (uta_mhdr_t *) nn_allocmsg( mlen, 0 )) == NULL ) {				// this will be released on send, so DO NOT free
		fprintf( stderr, "[CRIT] rmr_alloc_zc: cannot get memory for zero copy buffer: %d\n", errno );
		exit( 1 );
	}

	memset( msg->header, 0, sizeof( uta_mhdr_t ) );			// must ensure that header portion of tpbuf is 0
	msg->tp_buf = msg->header;
	hdr = (uta_mhdr_t *) msg->header;
	hdr->rmr_ver = htonl( RMR_MSG_VER );								// current version
	hdr->sub_id = htonl( UNSET_SUBID );
	SET_HDR_LEN( hdr );
	SET_HDR_TR_LEN( hdr, tr_len );							// set the actual length used
	//SET_HDR_D1_LEN( hdr, ctx->d1_len );					// moot until we actually need these data areas
	//SET_HDR_D2_LEN( hdr, ctx->d1_len );

	msg->len = 0;											// length of data in the payload
	msg->alloc_len = mlen;									// length of allocated payload
	msg->sub_id = UNSET_SUBID;
	msg->mtype = UNSET_MSGTYPE;
	msg->payload = PAYLOAD_ADDR( hdr );						// point at the payload in transport
	msg->xaction = ((uta_mhdr_t *)msg->header)->xid;		// point at transaction id in header area
	msg->state = state;										// fill in caller's state (likely the state of the last operation)
	msg->flags |= MFL_ZEROCOPY;								// this is a zerocopy sendable message
	strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, ctx->my_name, RMR_MAX_SID );

	if( DEBUG > 1 ) fprintf( stderr, "[DBUG] alloc_zcmsg mlen = %d size=%d mpl=%d flags=%02x %p m=%p @%p\n", mlen, size, ctx->max_plen, msg->flags, &msg->flags, msg, msg->header );

	return msg;
}

/*
	This will clone a message into a new zero copy buffer and return the cloned message.
*/
static inline rmr_mbuf_t* clone_msg( rmr_mbuf_t* old_msg  ) {
	rmr_mbuf_t* nm;			// new message buffer
	int	mlen;

	if( old_msg == NULL ) {
		return NULL;
	}

	nm = (rmr_mbuf_t *) malloc( sizeof *nm );
	if( nm == NULL ) {
		fprintf( stderr, "[CRIT] rmr_clone: cannot get memory for message buffer\n" );
		exit( 1 );
	}
	memset( nm, 0, sizeof( *nm ) );

	mlen = old_msg->alloc_len;							// length allocated before
	if( (nm->header = (uta_mhdr_t *) nn_allocmsg( mlen, 0 )) == NULL ) {				// this will be released on send, so DO NOT free
		fprintf( stderr, "[CRIT] rmr_clone: cannot get memory for zero copy buffer: %d\n", errno );
		exit( 1 );
	}

	memcpy( nm->header, old_msg->header, RMR_HDR_LEN( old_msg->header ) );     // copy complete header, trace and other data

	nm->mtype = old_msg->mtype;
	nm->sub_id = old_msg->sub_id;
	nm->len = old_msg->len;									// length of data in the payload
	nm->alloc_len = mlen;									// length of allocated payload
	nm->payload = PAYLOAD_ADDR( nm->header );				// reference the payload
	nm->xaction = ((uta_mhdr_t *)nm->header)->xid;			// point at transaction id in header area
	nm->state = old_msg->state;								// fill in caller's state (likely the state of the last operation)
	nm->flags |= MFL_ZEROCOPY;								// this is a zerocopy sendable message
	memcpy( nm->payload, old_msg->payload, old_msg->len );

	return nm;
}

static inline rmr_mbuf_t* realloc_msg( rmr_mbuf_t* old_msg, int tr_len  ) {
	rmr_mbuf_t* nm;			// new message buffer
	size_t	mlen;
	int state;
	uta_mhdr_t* hdr;
	uta_v1mhdr_t* v1hdr;
	int	tr_old_len;			// tr size in new buffer


	nm = (rmr_mbuf_t *) malloc( sizeof *nm );
	if( nm == NULL ) {
		fprintf( stderr, "[CRI] rmr_clone: cannot get memory for message buffer\n" );
		exit( 1 );
	}
	memset( nm, 0, sizeof( *nm ) );

	hdr = old_msg->header;
	tr_old_len = RMR_TR_LEN( hdr );				// bytes in old header for trace

	mlen = old_msg->alloc_len + (tr_len - tr_old_len);							// new length with trace adjustment
	if( DEBUG ) fprintf( stderr, "tr_realloc old size=%d new size=%d new tr_len=%d\n", (int) old_msg->alloc_len, (int) mlen, (int) tr_len );
	if( (nm->header = (uta_mhdr_t *) nn_allocmsg( mlen, 0 )) == NULL ) {				// this will be released on send, so DO NOT free
		fprintf( stderr, "[CRIT] rmr_realloc: cannot get memory for zero copy buffer: %d\n", errno );
		exit( 1 );
	}

	nm->tp_buf = nm->header;								// in nano both are the same
	v1hdr = (uta_v1mhdr_t *) old_msg->header;				// v1 will work to dig header out of any version
	switch( ntohl( v1hdr->rmr_ver ) ) {
		case 1:
			memcpy( v1hdr, old_msg->header, sizeof( *v1hdr ) );	 	// copy complete header
			nm->payload = (void *) v1hdr + sizeof( *v1hdr );
			break;

		default:											// current message always caught  here
			hdr = nm->header;
			memcpy( hdr, old_msg->header, sizeof( uta_mhdr_t ) );	 	// ONLY copy the header portion; trace and data offsets might have changed
			if( RMR_D1_LEN( hdr )  ) {
				memcpy( DATA1_ADDR( hdr ), DATA1_ADDR( old_msg->header ), RMR_D1_LEN( hdr ) );	 	// copy data1 and data2 if necessary

			}
			if( RMR_D2_LEN( hdr )  ) {
				memcpy( DATA2_ADDR( hdr ), DATA2_ADDR( old_msg->header ), RMR_D2_LEN( hdr ) );	 	// copy data1 and data2 if necessary
			}

			SET_HDR_TR_LEN( hdr, tr_len );										// len MUST be set before pointing payload
			nm->payload = PAYLOAD_ADDR( hdr );									// reference user payload
			break;
	}

	// --- these are all version agnostic -----------------------------------
	nm->mtype = old_msg->mtype;
	nm->sub_id = old_msg->sub_id;
	nm->len = old_msg->len;									// length of data in the payload
	nm->alloc_len = mlen;									// length of allocated payload

	nm->xaction = hdr->xid;									// reference xaction
	nm->state = old_msg->state;								// fill in caller's state (likely the state of the last operation)
	nm->flags = old_msg->flags | MFL_ZEROCOPY;				// this is a zerocopy sendable message
	memcpy( nm->payload, old_msg->payload, old_msg->len );

	return nm;
}

/*
	This is the receive work horse used by the outer layer receive functions.
	It waits for a message to be received on our listen socket. If old msg
	is passed in, the we assume we can use it instead of allocating a new
	one, else a new block of memory is allocated.

	This allocates a zero copy message so that if the user wishes to call
	uta_rts_msg() the send is zero copy.
*/
static rmr_mbuf_t* rcv_msg( uta_ctx_t* ctx, rmr_mbuf_t* old_msg ) {
	int nn_sock;				// endpoint socket for send
	int state;
	rmr_mbuf_t*	msg = NULL;		// msg received
	uta_mhdr_t* hdr;

	if( old_msg ) {
		msg = old_msg;
	} else {
		msg = alloc_zcmsg( ctx, NULL, RMR_MAX_RCV_BYTES, RMR_OK, DEF_TR_LEN );			// will abort on failure, no need to check
	}

	msg->state = nn_recv( ctx->nn_sock, msg->header, msg->alloc_len, NO_FLAGS );		// total space (header + payload len) allocated
	if( msg->state > (int) sizeof( uta_mhdr_t ) ) {						// we need more than just a header here
		hdr = (uta_mhdr_t *) msg->header;
		msg->len = ntohl( hdr->plen );						// length of data in the payload (likely < payload size)
		if( msg->len > msg->state - RMR_HDR_LEN( hdr ) ) {
			msg->state = RMR_ERR_TRUNC;
			msg->len = msg->state - RMR_HDR_LEN( hdr );
		}
		msg->mtype = ntohl( hdr->mtype );								// capture and convert from network order to local order
		msg->sub_id = ntohl( hdr->sub_id );								// capture and convert from network order to local order
		msg->state = RMR_OK;
		msg->flags |= MFL_ADDSRC;										// turn on so if user app tries to send this buffer we reset src
		msg->payload = PAYLOAD_ADDR( msg->header );
		msg->xaction = &hdr->xid[0];							// provide user with ref to fixed space xaction id
		if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rcv_msg: got something: type=%d state=%d len=%d diff=%ld\n",
				msg->mtype, msg->state, msg->len,  msg->payload - (unsigned char *) msg->header );
	} else {
		msg->len = 0;
		msg->state = RMR_ERR_EMPTY;
	}

	return msg;
}


/*
	Receives a 'raw' message from a non-RMr sender (no header expected). The returned
	message buffer cannot be used to send, and the length information may or may
	not be correct (it is set to the length received which might be more than the
	bytes actually in the payload).
*/
static void* rcv_payload( uta_ctx_t* ctx, rmr_mbuf_t* old_msg ) {
	int nn_sock;				// endpoint socket for send
	int state;
	rmr_mbuf_t*	msg = NULL;		// msg received

	if( old_msg ) {
		msg = old_msg;
	} else {
		msg = alloc_zcmsg( ctx, NULL, RMR_MAX_RCV_BYTES, RMR_OK, DEF_TR_LEN );			// will abort on failure, no need to check
	}

	msg->state = nn_recv( ctx->nn_sock, msg->header, msg->alloc_len, NO_FLAGS );		// read and state will be length
	if( msg->state >= 0 ) {
		msg->xaction = NULL;
		msg->mtype = UNSET_MSGTYPE;
		msg->sub_id = UNSET_SUBID;
		msg->len = msg->state;										// no header; len is the entire thing received
		msg->state = RMR_OK;
		msg->flags = MFL_RAW;										// prevent any sending of this headerless buffer
		msg->payload = msg->header;
		if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rcv_payload: got something: type=%d state=%d len=%d\n", msg->mtype, msg->state, msg->len );
	} else {
		msg->len = 0;
		msg->state = RMR_ERR_EMPTY;
		msg->payload = NULL;
		msg->xaction = NULL;
		msg->mtype = UNSET_MSGTYPE;
		msg->sub_id = UNSET_SUBID;
	}

	return msg;
}

/*
	This does the hard work of actually sending the message to the given socket. On success,
	a new message struct is returned. On error, the original msg is returned with the state
	set to a reasonable value. If the message being sent as MFL_NOALLOC set, then a new
	buffer will not be allocated and returned (mostly for call() interal processing since
	the return message from call() is a received buffer, not a new one).

	Called by rmr_send_msg() and rmr_rts_msg().
*/
static rmr_mbuf_t* send_msg( uta_ctx_t* ctx, rmr_mbuf_t* msg, int nn_sock ) {
	int state;
	uta_mhdr_t*	hdr;
	int	tr_len;					// length from the message being sent (must snarf before send to use after send)

	// future: ensure that application did not overrun the XID buffer; last byte must be 0

	//fprintf( stderr, ">>>>>> sending to %d %d\n", nn_sock, msg->mtype );
	hdr = (uta_mhdr_t *) msg->header;
	hdr->mtype = htonl( msg->mtype );								// stash type/len/sub-id in network byte order for transport
	hdr->sub_id = htonl( msg->sub_id );
	hdr->plen = htonl( msg->len );

	if( msg->flags & MFL_ADDSRC ) {									// buffer was allocated as a receive buffer; must add our source
		strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, ctx->my_name, RMR_MAX_SID );					// must overlay the source to be ours
	}

	tr_len = RMR_TR_LEN( hdr );
	if( msg->flags & MFL_ZEROCOPY ) {									// faster sending with zcopy buffer
		if( (state = nn_send( nn_sock, &msg->header, NN_MSG, NN_DONTWAIT )) < 0 ) {
			msg->state = state;
		} else {
			msg->header = NULL;											// nano frees; don't risk accessing later by mistake
		}
	} else {
		if( (state = nn_send( nn_sock, msg->header, sizeof( uta_mhdr_t ) + msg->len, NN_DONTWAIT )) < 0 ) {
			msg->state = state;
		}
	}

	// future:  if nano sends bytes, but less than mlen, then what to do?
	if( msg->state >= 0 ) {										// successful send
		if( !(msg->flags & MFL_NOALLOC) ) {						// if noalloc is set, then caller doesn't want a new buffer
			return alloc_zcmsg( ctx, msg, 0, RMR_OK, tr_len );	// preallocate a zero-copy buffer and return msg (with same trace len as sent buffer)
		} else {
			rmr_free_msg( msg );						// not wanting a meessage back, trash this one
			return NULL;
		}
	} else {											// send failed -- return original message
		if( errno == EAGAIN ) {
			msg->state = RMR_ERR_RETRY;					// some wrappers can't see errno, make this obvious
		} else {
			msg->state = RMR_ERR_SENDFAILED;					// errno will have nano reason
		}
		if( DEBUG ) fprintf( stderr, "[DBUG] send failed: %s\n", strerror( errno ) );
	}

	return msg;
}


/*
	A generic wrapper to the real send to keep wormhole stuff agnostic.
	We assume the wormhole function vetted the buffer so we don't have to.
*/
static rmr_mbuf_t* send2ep( uta_ctx_t* ctx, endpoint_t* ep, rmr_mbuf_t* msg ) {
	return send_msg( ctx, msg, ep->nn_sock );
}

#endif