--- /dev/null
+// :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