/*
Mnemonic: sr_si_static.c
Abstract: These are static send/receive primatives which (sadly)
- differ based on the underlying protocol (nng vs nanomsg).
+ differ based on the underlying protocol (nng vs SI95).
Split from rmr_nng.c for easier wormhole support.
Author: E. Scott Daniels
#ifndef _sr_si_static_c
#define _sr_si_static_c
-#include <nng/nng.h>
-#include <nng/protocol/pubsub0/pub.h>
-#include <nng/protocol/pubsub0/sub.h>
-#include <nng/protocol/pipeline0/push.h>
-#include <nng/protocol/pipeline0/pull.h>
-
-
static void dump_40( char *p, char* label ) {
int i;
if( msg == NULL && (msg = (rmr_mbuf_t *) uta_ring_extract( ctx->zcb_mring )) == NULL ) {
msg = (rmr_mbuf_t *) malloc( sizeof *msg );
if( msg == NULL ) {
- fprintf( stderr, "[CRI] rmr_alloc_zc: cannot get memory for message\n" );
+ rmr_vlog( RMR_VL_CRIT, "rmr_alloc_zc: cannot get memory for message\n" );
return NULL; // we used to exit -- that seems wrong
}
memset( msg, 0, sizeof( *msg ) ); // tp_buffer will be allocated below
}
}
+ msg->rts_fd = -1; // must force to be invalid; not a received message that can be returned
if( !msg->alloc_len && (msg->tp_buf = (void *) malloc( mlen )) == NULL ) {
- fprintf( stderr, "[CRI] rmr_alloc_zc: cannot get memory for zero copy buffer: %d bytes\n", (int) mlen );
+ rmr_vlog( RMR_VL_CRIT, "rmr_alloc_zc: cannot get memory for zero copy buffer: %d bytes\n", (int) mlen );
abort( ); // toss out a core file for this
}
strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, ctx->my_name, RMR_MAX_SRC );
strncpy( (char *) ((uta_mhdr_t *)msg->header)->srcip, ctx->my_ip, RMR_MAX_SRC );
- if( DEBUG > 1 ) fprintf( stderr, "[DBUG] alloc_zcmsg mlen=%ld size=%d mpl=%d flags=%02x\n", (long) mlen, size, ctx->max_plen, msg->flags );
+ if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "alloc_zcmsg mlen=%ld size=%d mpl=%d flags=%02x\n", (long) mlen, size, ctx->max_plen, msg->flags );
return msg;
}
}
} else {
if( (msg = (rmr_mbuf_t *) malloc( sizeof *msg )) == NULL ) {
- fprintf( stderr, "[CRI] rmr_alloc_mbuf: cannot get memory for message\n" );
+ rmr_vlog( RMR_VL_CRIT, "rmr_alloc_mbuf: cannot get memory for message\n" );
return NULL; // this used to exit, but that seems wrong
}
}
nm = (rmr_mbuf_t *) malloc( sizeof *nm );
if( nm == NULL ) {
- fprintf( stderr, "[CRI] rmr_clone: cannot get memory for message buffer\n" );
+ rmr_vlog( RMR_VL_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->tp_buf = (void *) malloc( sizeof( char ) * (mlen + TP_HDR_LEN) )) == NULL ) {
- fprintf( stderr, "[CRI] rmr_si_clone: cannot get memory for zero copy buffer: %d\n", (int) mlen );
+ rmr_vlog( RMR_VL_CRIT, "rmr_si_clone: cannot get memory for zero copy buffer: %d\n", (int) mlen );
abort();
}
nm = (rmr_mbuf_t *) malloc( sizeof *nm );
if( nm == NULL ) {
- fprintf( stderr, "[CRI] rmr_clone: cannot get memory for message buffer\n" );
+ rmr_vlog( RMR_VL_CRIT, "rmr_clone: cannot get memory for message buffer\n" );
exit( 1 );
}
memset( nm, 0, sizeof( *nm ) );
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, "[DBUG] tr_realloc old size=%d new size=%d new tr_len=%d\n", (int) old_msg->alloc_len, (int) mlen, (int) tr_len );
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "tr_realloc old size=%d new size=%d new tr_len=%d\n", (int) old_msg->alloc_len, (int) mlen, (int) tr_len );
tpb_len = mlen + TP_HDR_LEN;
if( (nm->tp_buf = (void *) malloc( tpb_len)) == NULL ) {
- fprintf( stderr, "[CRI] rmr_clone: cannot get memory for zero copy buffer: %d\n", ENOMEM );
+ rmr_vlog( RMR_VL_CRIT, "rmr_clone: cannot get memory for zero copy buffer: %d\n", ENOMEM );
exit( 1 );
}
memset( nm->tp_buf, 0, tpb_len );
return nm;
}
+/*
+ Realloc the message such that the payload is at least payload_len bytes.
+ The clone and copy options affect what portion of the original payload is copied to
+ the reallocated message, and whether or not the original payload is lost after the
+ reallocation process has finished.
+
+ copy == true
+ The entire payload from the original message will be coppied to the reallocated
+ payload.
+
+ copy == false
+ Only the header (preserving return to sender information, message type, etc)
+ is preserved after reallocation; the payload used lengrh is set to 0 and the
+ payload is NOT initialised/cleared.
+
+ clone == true
+ The orignal message is preserved and a completely new message buffer and payload
+ are allocated (even if the size given is the same). A pointer to the new message
+ buffer is returned and it is the user application's responsibility to manage the
+ old buffer (e.g. free when not needed).
+
+ clone == false
+ The old payload will be lost after reallocation. The message buffer pointer which
+ is returned will likely reference the same structure (don't depend on that).
+
+
+ CAUTION:
+ If the message is not a message which was received, the mtype, sub-id, length values in the
+ RMR header in the allocated transport buffer will NOT be accurate and will cause the resulting
+ mbuffer information for mtype and subid to be reset even when copy is true. To avoid silently
+ resetting information in the mbuffer, this funciton will reset the mbuf values from the current
+ settings and NOT from the copied RMR header in transport buffer.
+*/
+static inline rmr_mbuf_t* realloc_payload( rmr_mbuf_t* old_msg, int payload_len, int copy, int clone ) {
+ rmr_mbuf_t* nm = NULL; // new message buffer when cloning
+ size_t mlen;
+ uta_mhdr_t* omhdr; // old message header
+ int tr_old_len; // tr size in new buffer
+ int old_psize = 0; // size of payload in the message passed in (alloc size - tp header and rmr header lengths)
+ int hdr_len = 0; // length of RMR and transport headers in old msg
+ void* old_tp_buf; // pointer to the old tp buffer
+ int free_tp = 1; // free the transport buffer (old) when done (when not cloning)
+ int old_mt; // msg type and sub-id from the message passed in
+ int old_sid;
+ int old_len;
+ int old_rfd; // rts file descriptor from old message
+
+ if( old_msg == NULL || payload_len <= 0 ) {
+ errno = EINVAL;
+ return NULL;
+ }
+
+ old_mt = old_msg->mtype; // preserve mbuf info
+ old_sid = old_msg->sub_id;
+ old_len = old_msg->len;
+ old_rfd = old_msg->rts_fd;
+
+ old_psize = old_msg->alloc_len - (RMR_HDR_LEN( old_msg->header ) + TP_HDR_LEN); // user payload size in orig message
+
+ if( !clone && payload_len <= old_psize ) { // not cloning and old is large enough; nothing to do
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rmr_realloc_payload: old msg payload larger than requested: cur=%d need=%d\n", old_psize, payload_len );
+ return old_msg;
+ }
+
+ hdr_len = RMR_HDR_LEN( old_msg->header ) + TP_HDR_LEN; // with SI we manage the transport header; must include in len
+ old_tp_buf = old_msg->tp_buf;
+
+ if( clone ) {
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rmr_realloc_payload: cloning message\n" );
+ free_tp = 0;
+
+ nm = (rmr_mbuf_t *) malloc( sizeof( *nm ) );
+ if( nm == NULL ) {
+ rmr_vlog( RMR_VL_CRIT, "rmr_realloc_payload: cannot get memory for message buffer. bytes requested: %d\n", (int) sizeof(*nm) );
+ return NULL;
+ }
+ memset( nm, 0, sizeof( *nm ) );
+ nm->rts_fd = old_rfd; // this is managed only in the mbuf; dup now
+ } else {
+ nm = old_msg;
+ }
+
+ omhdr = old_msg->header;
+ mlen = hdr_len + (payload_len > old_psize ? payload_len : old_psize); // must have larger in case copy is true
+
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "reallocate for payload increase. new message size: %d\n", (int) mlen );
+ if( (nm->tp_buf = (char *) malloc( sizeof( char ) * mlen )) == NULL ) {
+ rmr_vlog( RMR_VL_CRIT, "rmr_realloc_payload: cannot get memory for zero copy buffer. bytes requested: %d\n", (int) mlen );
+ return NULL;
+ }
+
+ nm->header = ((char *) nm->tp_buf) + TP_HDR_LEN; // point at the new header and copy from old
+ SET_HDR_LEN( nm->header );
+
+ if( copy ) { // if we need to copy the old payload too
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rmr_realloc_payload: copy payload into new message: %d bytes\n", old_psize );
+ memcpy( nm->header, omhdr, sizeof( char ) * (old_psize + RMR_HDR_LEN( omhdr )) );
+ } else { // just need to copy header
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rmr_realloc_payload: copy only header into new message: %d bytes\n", RMR_HDR_LEN( nm->header ) );
+ memcpy( nm->header, omhdr, sizeof( char ) * RMR_HDR_LEN( omhdr ) );
+ }
+
+ ref_tpbuf( nm, mlen ); // set payload and other pointers in the message to the new tp buffer
+
+ if( !copy ) {
+ nm->mtype = -1; // didn't copy payload, so mtype, sub-id, and rts fd are invalid
+ nm->sub_id = -1;
+ nm->len = 0; // and len is 0
+ } else {
+ nm->len = old_len; // we must force these to avoid losing info if msg wasn't a received message
+ nm->mtype = old_mt;
+ nm->sub_id = old_sid;
+ }
+
+ if( free_tp ) {
+ free( old_tp_buf ); // we did not clone, so free b/c no references
+ }
+
+ return nm;
+}
+
/*
For SI95 based transport all receives are driven through the threaded
ring and thus this function should NOT be called. If it is we will panic
static void* rcv_payload( uta_ctx_t* ctx, rmr_mbuf_t* old_msg ) {
return NULL;
/*
-FIXME: not implemented yet
+FIXME: do we need this in the SI world? The only user was the route table collector
int state;
rmr_mbuf_t* msg = NULL; // msg received
size_t rsize; // nng needs to write back the size received... grrr
msg->payload = msg->header; // payload is the whole thing; no header
msg->xaction = NULL;
- if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rcv_payload: got something: type=%d state=%d len=%d\n", msg->mtype, msg->state, msg->len );
+ if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "rcv_payload: got something: type=%d state=%d len=%d\n", msg->mtype, msg->state, msg->len );
return msg;
*/
tot_len = msg->len + PAYLOAD_OFFSET( hdr ) + TP_HDR_LEN; // we only send what was used + header lengths
*((int*) msg->tp_buf) = tot_len;
- if( DEBUG > 1 ) fprintf( stderr, "[DEBUG] send_msg: ending %d (%x) bytes usr_len=%d alloc=%d retries=%d\n", tot_len, tot_len, msg->len, msg->alloc_len, retries );
+ if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "send_msg: ending %d (%x) bytes usr_len=%d alloc=%d retries=%d\n", tot_len, tot_len, msg->len, msg->alloc_len, retries );
if( DEBUG > 2 ) dump_40( msg->tp_buf, "sending" );
if( (state = SIsendt( ctx->si_ctx, nn_sock, msg->tp_buf, tot_len )) != SI_OK ) {
- if( DEBUG > 1 ) fprintf( stderr, "[DBUG] send_msg: error!! sent state=%d\n", state );
+ if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "send_msg: error!! sent state=%d\n", state );
msg->state = state;
if( retries > 0 && state == SI_ERR_BLOCKED ) {
if( --spin_retries <= 0 ) { // don't give up the processor if we don't have to
state = 0; // don't loop
}
} else {
- if( DEBUG > 2 ) fprintf( stderr, "[DBUG] sent OK state=%d\n", state );
+ if( DEBUG > 2 ) rmr_vlog( RMR_VL_DEBUG, "sent OK state=%d\n", state );
state = 0;
msg->state = RMR_OK;
hdr = NULL;
msg->state = RMR_ERR_SENDFAILED;
}
- if( DEBUG ) fprintf( stderr, "[DBUG] send failed: %d %s\n", (int) msg->state, strerror( msg->state ) );
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "send failed: %d %s\n", (int) msg->state, strerror( msg->state ) );
}
return msg;
}
if( (rte = uta_get_rte( ctx->rtable, msg->sub_id, msg->mtype, TRUE )) == NULL ) { // find the entry which matches subid/type allow fallback to type only key
- if( ctx->flags & CTXFL_WARN ) {
- fprintf( stderr, "[WARN] no route table entry for mtype=%d sub_id=%d\n", msg->mtype, msg->sub_id );
- }
+ rmr_vlog( RMR_VL_WARN, "no route table entry for mtype=%d sub_id=%d\n", msg->mtype, msg->sub_id );
msg->state = RMR_ERR_NOENDPT;
errno = ENXIO; // must ensure it's not eagain
msg->tp_state = errno;
while( send_again ) {
sock_ok = uta_epsock_rr( rte, group, &send_again, &nn_sock, &ep, ctx->si_ctx ); // select endpt from rr group and set again if more groups
- if( DEBUG ) fprintf( stderr, "[DBUG] mtosend_msg: flgs=0x%04x type=%d again=%d group=%d len=%d sock_ok=%d\n",
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "mtosend_msg: flgs=0x%04x type=%d again=%d group=%d len=%d sock_ok=%d\n",
msg->flags, msg->mtype, send_again, group, msg->len, sock_ok );
group++;
msg->state = RMR_ERR_SENDFAILED;
errno = ENOMEM;
msg->tp_state = errno;
- if( ctx->flags & CTXFL_WARN ) {
- fprintf( stderr, "[WARN] unable to clone message for multiple rr-group send\n" );
- }
+ rmr_vlog( RMR_VL_WARN, "unable to clone message for multiple rr-group send\n" );
return msg;
}
- if( DEBUG ) fprintf( stderr, "[DBUG] msg cloned: type=%d len=%d\n", msg->mtype, msg->len );
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "msg cloned: type=%d len=%d\n", msg->mtype, msg->len );
msg->flags |= MFL_NOALLOC; // keep send from allocating a new message; we have a clone to use
msg = send_msg( ctx, msg, nn_sock, max_to ); // do the hard work, msg should be nil on success
msg = send_msg( ctx, msg, nn_sock, max_to ); // send the last, and allocate a new buffer; drops the clone if it was
if( DEBUG ) {
if( msg == NULL ) {
- fprintf( stderr, "[DBUG] mtosend_msg: send returned nil message!\n" );
+ rmr_vlog( RMR_VL_DEBUG, "mtosend_msg: send returned nil message!\n" );
}
}
}
}
}
} else {
-/*
- if( ctx->flags & CTXFL_WARN ) {
- fprintf( stderr, "[WARN] invalid socket for rte, setting no endpoint err: mtype=%d sub_id=%d\n", msg->mtype, msg->sub_id );
- }
-*/
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "invalid socket for rte, setting no endpoint err: mtype=%d sub_id=%d\n", msg->mtype, msg->sub_id );
msg->state = RMR_ERR_NOENDPT;
errno = ENXIO;
}
msg->state = RMR_OK;
}
- if( DEBUG ) fprintf( stderr, "[DBUG] final send stats: ok=%d group=%d state=%d\n\n", ok_sends, group, msg->state );
+ if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "final send stats: ok=%d group=%d state=%d\n", ok_sends, group, msg->state );
msg->tp_state = errno;
}
Code Review / ric-plt / lib / rmr.git / blobdiff