BodgeMaster
/
NORM-mirror
mirror of https://github.com/USNavalResearchLaboratory/norm.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

v1.3b8

pull/2/head
Jeff Weston 2019-09-11 11:54:39 -04:00
parent 3dd25deec9
commit 098570e3ad
31 changed files with 2494 additions and 346 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

290
NormDeveloperGuide.html
View File

@ -6,7 +6,7 @@
<META NAME="GENERATOR" CONTENT="NeoOffice/J 1.1 (Unix)">
<META NAME="AUTHOR" CONTENT="Robert Adamson">
<META NAME="CREATED" CONTENT="20040824;14180000">
<META NAME="CHANGED" CONTENT="20051028;14285700">
<META NAME="CHANGED" CONTENT="20060512;16004100">
<META NAME="DESCRIPTION" CONTENT="NORM Object Functions 42 NORM Object Functions NORM_OBJECT_STREAM NORM Sender Functions 19 NORM Sender Functions ">
<META NAME="KEYWORDS" CONTENT="norm object value application function ">
<STYLE>
@ -52,7 +52,7 @@
<TD WIDTH=371>
<P ALIGN=LEFT STYLE="margin-left: 0.86in; margin-top: 0.17in; font-style: normal">
<FONT FACE="Helvetica, sans-serif"><FONT SIZE=4 STYLE="font-size: 16pt"><B><FONT SIZE=6 STYLE="font-size: 28pt">NORM<BR>Developer’s
<BR>Guide<BR></FONT><FONT SIZE=4>(version 1.3b7)</FONT></B></FONT></FONT></P>
<BR>Guide<BR></FONT><FONT SIZE=4>(version 1.3b8)</FONT></B></FONT></FONT></P>
</TD>
</TR>
</TABLE>
@ -163,6 +163,7 @@ of Contents</A></H1>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormSetTransmitRateBounds()">NormSetTransmitRateBounds()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormSetTransmitCacheBounds()">NormSetTransmitCacheBounds()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormSetAutoParity()">NormSetAutoParity()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormGetGrttEstimate()">NormGetGrttEstimate()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormSetGrttEstimate()">NormSetGrttEstimate()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormSetGrttMax()">NormSetGrttMax()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormSetGrttProbingMode()">NormSetGrttProbingMode()</A></FONT></P>
@ -171,6 +172,7 @@ of Contents</A></H1>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormSetGroupSize()">NormSetGroupSize()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormFileEnqueue()">NormFileEnqueue()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormDataEnqueue()">NormDataEnqueue()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormRequeueObject()">NormRequeueObject()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormStreamOpen()">NormStreamOpen()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormStreamClose()">NormStreamClose()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormStreamWrite()">NormStreamWrite()</A></FONT></P>
@ -216,6 +218,7 @@ of Contents</A></H1>
<P STYLE="margin-top: 0.17in; font-style: normal"><FONT SIZE=2><B><A HREF="#NORM Node Functions">NORM Node Functions</A></B></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormNodeGetId()">NormNodeGetId()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormNodeGetAddress()">NormNodeGetAddress()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormNodeGetGrtt()">NormNodeGetGrtt()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormNodeRetain()">NormNodeRetain()</A></FONT></P>
<P STYLE="margin-left: 0.17in; margin-top: 0in; font-style: normal"><FONT SIZE=2><A HREF="#NormNodeRelease()">NormNodeRelease()</A></FONT></P>
</DIV>
@ -1039,7 +1042,7 @@ function call.</P>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_LOCAL_SERVER_CLOSED</I></FONT></P>
<FONT FACE="Courier"><I>NORM_LOCAL_SENDER_CLOSED</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
@ -1048,6 +1051,40 @@ function call.</P>
the indicated &quot;session&quot; (see <FONT FACE="Courier"><A HREF="#NormStopSender()">NormStopSender()</A></FONT>).</P>
</TD>
</TR>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_CC_ACTIVE</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
This event indicates that congestion control feedback from
receivers has begun to be received (<I>This also implies that
receivers in the group are actually present and can be used as a
cue to begin data transmission.</I>). Note that congestion
control must be enabled (see <FONT FACE="Courier"><A HREF="#NormSetCongestionControl()">NormSetCongestionControl()</A></FONT>)
for this event to be posted. Congestion control feedback can be
assumed to be received until a <FONT FACE="Courier"><I>NORM_CC_INACTIVE</I></FONT>
event is posted.</P>
</TD>
</TR>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_CC_INACTIVE</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
This event indicates there has been no recent congestion control
feedback received from the receiver set and that the local NORM
sender has reached its minimum transmit rate. Applications may
wish to refrain from new data transmission until a <FONT FACE="Courier"><I>NORM_CC_ACTIVE</I></FONT>
event is posted. This notification is only posted when congestion
control operation is enabled (see <FONT FACE="Courier"><A HREF="#NormSetCongestionControl()">NormSetCongestionControl()</A></FONT>)
<I>and</I> a previous <FONT FACE="Courier"><I>NORM_CC_ACTIVE</I></FONT>
event has occurred.</P>
</TD>
</TR>
<TR>
<TD COLSPAN=2 WIDTH=576 VALIGN=TOP>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
@ -1057,32 +1094,32 @@ function call.</P>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_REMOTE_SERVER_NEW</I></FONT></P>
<FONT FACE="Courier"><I>NORM_REMOTE_SENDER_NEW</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
This notification is posted when a receiver first receives
messages from a specific remote NORM server. This marks the
beginning of the interval during which the application may
reference the provided &quot;node&quot; handle (<FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT>).</P>
This event is posted when a receiver first receives messages from
a specific remote NORM sender. This marks the beginning of the
interval during which the application may reference the provided
&quot;node&quot; handle (<FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT>).</P>
</TD>
</TR>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_REMOTE_SERVER_ACTIVE</I></FONT></P>
<FONT FACE="Courier"><I>NORM_REMOTE_SENDER_ACTIVE</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid"><A NAME="OLE_LINK5"></A><A NAME="OLE_LINK4"></A>
This event is posted when a previously inactive (or new) remote
server is detected operating as an active sender within the
sender is detected operating as an active sender within the
session.</P>
</TD>
</TR>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_REMOTE_SERVER_INACTIVE</I></FONT></P>
<FONT FACE="Courier"><I>NORM_REMOTE_SENDER_INACTIVE</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
@ -1095,7 +1132,7 @@ function call.</P>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_REMOTE_SERVER_PURGED</I></FONT></P>
<FONT FACE="Courier"><I>NORM_REMOTE_SENDER_PURGED</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
@ -1172,6 +1209,22 @@ function call.</P>
<B>Miscellaneous Notification Event Types</B></P>
</TD>
</TR>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
<FONT FACE="Courier"><I>NORM_GRTT_UPDATED</I></FONT></P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
This notification indicates that either the local sender estimate
of GRTT has changed, or that a remote sender's estimate of GRTT
has changed. The &quot;sender&quot; member of the <FONT FACE="Courier"><A HREF="#NormEvent">NormEvent</A></FONT>
is set to <FONT FACE="Courier"><I>NORM_NODE_INVALID</I></FONT> if
the local sender's GRTT estimate has changed or to the
<FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT> of the remote sender
that has updated its estimate of GRTT.</P>
</TD>
</TR>
<TR VALIGN=TOP>
<TD WIDTH=269>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
@ -1179,8 +1232,8 @@ function call.</P>
</TD>
<TD WIDTH=293>
<P CLASS="western" STYLE="font-style: normal; page-break-inside: avoid">
This <A HREF="#NormEventType">NormEventType</A> indicates an invalid or &quot;null&quot;
notification which should be ignored.</P>
This <FONT FACE="Courier"><A HREF="#NormEventType">NormEventType</A></FONT> indicates an
invalid or &quot;null&quot; notification which should be ignored.</P>
</TD>
</TR>
</TABLE>
@ -1800,7 +1853,7 @@ without notifying the receiver set of its intention. However a
&quot;graceful shutdown&quot; option is provided to terminate sender
operation gracefully, notifying the receiver set its pending exit
with appropriate protocol messaging. A <I><A HREF="#NormEvent">NormEvent</A></I>,
<FONT FACE="Courier"><I>NORM_LOCAL_SERVER_CLOSED,</I></FONT> is
<FONT FACE="Courier"><I>NORM_LOCAL_SENDER_CLOSED,</I></FONT> is
dispatched when the graceful shutdown process has completed.</P>
<P CLASS="western" STYLE="font-style: normal"><I>(NOTE: The
&quot;</I><FONT FACE="Courier"><U><I>graceful</I></U></FONT><I>&quot;
@ -2032,6 +2085,30 @@ called for the given <FONT FACE="Courier"><U>sessionHandle</U></FONT>.</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">This function has no
return values.</P>
<H3 CLASS="functionheading-western"><A NAME="NormGetGrttEstimate()">NormGetGrttEstimate()</A></H3>
<H4 CLASS="western">Synopsis</H4>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>#include
&lt;normApi.h&gt;</FONT></FONT></P>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>double
NormGetGrttEstimate(<A HREF="#NormSessionHandle">NormSessionHandle</A>&nbsp;session);</FONT></FONT></P>
<H4 CLASS="western">Description</H4>
<P CLASS="western" STYLE="font-style: normal">This function returns
the sender's current estimate(in seconds) of group round-trip timing
(GRTT) for the given NORM <FONT FACE="Courier"><U>session</U></FONT>.
This function may be useful for applications to leverage for other
purposes the assessment of round-trip timing made by the NORM
protocol engine. For example, an application may scale its own
timeouts based on connectivity delays among participants in a NORM
session. Note that the <FONT FACE="Courier"><I>NORM_GRTT_UPDATED</I></FONT>
event is posted (see <FONT FACE="Courier"><A HREF="#NormGetNextEvent()">NormGetNextEvent()</A></FONT>)
by the NORM protocol engine to indicate when changes in the local
sender or remote senders' GRTT estimate occurs.</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">This function returns
the current sender group round-trip timing (GRTT) estimate (in units
of seconds). A value of <FONT FACE="Courier"><I>-1.0</I></FONT> is
returned if an invalid <FONT FACE="Courier"><U>session</U></FONT>
value is provided.</P>
<H3 CLASS="functionheading-western"><A NAME="NormSetGrttEstimate()">NormSetGrttEstimate()</A></H3>
<H4 CLASS="western">Synopsis</H4>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>#include
@ -2383,6 +2460,56 @@ previously enqueued NORM transmit objects are pending transmission.
Also the call will fail if the <FONT FACE="Courier"><U>infoLen</U></FONT>
parameter exceeds the local <I>NormSender</I> <FONT FACE="Courier"><U>segmentSize</U></FONT>
limit.</P>
<H3 CLASS="functionheading-western"><A NAME="NormRequeueObject()">NormRequeueObject()</A></H3>
<H4 CLASS="western">Synopsis</H4>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>#include
&lt;normApi.h&gt;</FONT></FONT></P>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>bool&nbsp;NormRequeueObject(<A HREF="#NormSessionHandle">NormSessionHandle</A>&nbsp;session,<BR>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<A HREF="#NormObjectHandle">NormObjectHandle</A>&nbsp;&nbsp;object);</FONT></FONT></P>
<H4 CLASS="western">Description</H4>
<P CLASS="western" STYLE="font-style: normal">This function allows
the application to resend (or reset transmission of) a
<FONT FACE="Courier"><I>NORM_OBJECT_FILE</I></FONT> or
<FONT FACE="Courier"><I>NORM_OBJECT_DATA</I></FONT> transmit object
that was previously enqueued for the indicated <FONT FACE="Courier"><U>session</U></FONT>.
This function is useful for applications sending to silent
(non-NACKing) receivers as it enables the receivers to take advantage
of multiple retransmissions of objects (including any auto-parity
set, see <FONT FACE="Courier"><A HREF="#NormSetAutoParity()">NormSetAutoParity()</A></FONT>) to more
robustly receive content. The <FONT FACE="Courier"><U>object</U></FONT>
parameter must be a valid transmit <A HREF="#NormObjectHandle">NormObjectHandle</A> that has not yet
been &quot;purged&quot; from the sender's transmit queue. Upon
success, the specified <FONT FACE="Courier"><U>object</U></FONT> will
be fully retransmitted using the same NORM object transport
identifier as was used on its initial transmission. This call may be
made at any time to restart transmission of a previously-enqueued
object, but the <FONT FACE="Courier"><I>NORM_TX_OBJECT_SENT</I></FONT>
or <FONT FACE="Courier"><I>NORM_TX_FLUSH_COMPLETED</I></FONT>
notifications can serve as good cues for an appropriate time to
resend an object. If multiple objects are re-queued, they will be
resent in order of their initial enqueuing.</P>
<P CLASS="western" STYLE="font-style: normal">The transmit cache
bounds set by <FONT FACE="Courier"><A HREF="#NormSetTransmitCacheBounds()">NormSetTransmitCacheBounds()</A></FONT>
determine the number of previously-sent objects retained in the
sender's transmit queue and that are thus eligible to be requeued for
retransmission. An object may be requeued via this call multiple
times, but each distinct requeue should be done after an indication
such as <FONT FACE="Courier"><I>NORM_TX_OBJECT_SENT</I></FONT> or
<FONT FACE="Courier"><I>NORM_TX_FLUSH_COMPLETED</I></FONT> for the
given object. Otherwise, the object will simply be reset from its
current transmission point to transmit from the beginning (i.e.
restart). Note that the object type <FONT FACE="Courier"><I>NORM_OBJECT_STREAM</I></FONT>
cannot currently be requeued.</P>
<P CLASS="western"><I>(TBD – should a &quot;numRepeats&quot;
parameter be added to this function?)</I></P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">A value of <FONT FACE="Courier"><I>true</I></FONT>
is returned upon success and a value of <FONT FACE="Courier"><I>false</I></FONT>
is returned upon failure. Possible reasons for failure include an
invalid object handle was provided (i.e. a non-transmit object or
transmit object that has been &quot;purged&quot; from the transmit
queue (see <FONT FACE="Courier"><I>NORM_TX_OBJECT_PURGED</I></FONT>))
or the provided <FONT FACE="Courier"><U>object</U></FONT> was of type
<FONT FACE="Courier"><I>NORM_OBJECT_STREAM</I></FONT>.</P>
<H3 CLASS="functionheading-western"><A NAME="NormStreamOpen()">NormStreamOpen()</A></H3>
<H4 CLASS="western">Synopsis</H4>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>#include
@ -3005,17 +3132,18 @@ long&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;bufferSpace);</FONT></FONT></P>
<P CLASS="western" STYLE="font-style: normal">This function initiates
the application's participation as a receiver within the <I>NormSession</I>
identified by the <FONT FACE="Courier"><U>session</U></FONT>
parameter. The receiver will respond with appropriate protocol
messages (unless <FONT FACE="Courier">NormSetSilentReceiver(</FONT><FONT FACE="Courier"><I>true</I></FONT><FONT FACE="Courier">)</FONT>
is invoked) and begin providing the application with
receiver-related <I><A HREF="#NormEvent">NormEvent</A></I> notification. The <FONT FACE="Courier"><U>bufferSpace</U></FONT>
parameter is used to set a limit on the amount of <FONT FACE="Courier"><U>bufferSpace</U></FONT>
parameter. The NORM protocol engine will begin providing the
application with receiver-related <I><A HREF="#NormEvent">NormEvent</A></I> notifications,
and, unless <FONT FACE="Courier">NormSetSilentReceiver(</FONT><FONT FACE="Courier"><I>true</I></FONT><FONT FACE="Courier">)</FONT>
is invoked, respond to senders with appropriate protocol messages.
The <FONT FACE="Courier"><U>bufferSpace</U></FONT> parameter is used
to set a limit on the amount of <FONT FACE="Courier"><U>bufferSpace</U></FONT>
allocated by the receiver per active <I>NormSender</I> within the
session. The appropriate <FONT FACE="Courier"><U>bufferSpace</U></FONT>
to use is a function of expected network <I>delay*bandwidth</I>
product and packet loss characteristics. A discussion of trade-offs
associated with NORM transmit and receiver buffer space selection is
provided later in this document. An insufficient bufferSpace
provided later in this document. An insufficient <FONT FACE="Courier"><U>bufferSpace</U></FONT>
allocation will result in potentially inefficient protocol operation,
even though reliable operation may be maintained. In some cases of a
large <I>delay*bandwidth</I> product and/or severe packet loss, a
@ -3023,8 +3151,7 @@ small <FONT FACE="Courier"><U>bufferSpace</U></FONT> allocation
(coupled with the lack of explicit flow control in NORM) may result
in the receiver &quot;re-syncing&quot; to the sender, resulting in
&quot;outages&quot; in the reliable transmissions from a sender (this
is similar to the conditions resulting in a TCP connection timeout
failure).
is analogous to a TCP connection timeout failure).
</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">A value of <FONT FACE="Courier"><I>true</I></FONT>
@ -3112,7 +3239,8 @@ specification was given.</P>
&lt;normApi.h&gt;</FONT></FONT></P>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>void
NormSetSilentReceiver(<A HREF="#NormSessionHandle">NormSessionHandle</A> session,
<BR>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;bool&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;silent);</FONT></FONT></P>
<BR>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;bool&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;silent,<BR>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;bool&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;lowDelay
= false);</FONT></FONT></P>
<H4 CLASS="western">Description</H4>
<P CLASS="western" STYLE="font-style: normal">This function provides
the option to configure a NORM receiver application as a &quot;silent
@ -3127,6 +3255,27 @@ needed for reliability and protocol operation. Silent receivers are
dependent upon proactive FEC transmission (see <FONT FACE="Courier"><A HREF="#NormSetAutoParity()">NormSetAutoParity()</A></FONT>)
or using repair information requested by other non-silent receivers
within the group to achieve reliable transfers.</P>
<P CLASS="western" STYLE="font-style: normal">The optional <FONT FACE="Courier"><U>lowDelay</U></FONT>
parameter is applicable only for reception of the <FONT FACE="Courier"><I>NORM_OBJECT_STREAM</I></FONT>
type. When <FONT FACE="Courier"><U>lowDelay</U></FONT> is set to a
value of <FONT FACE="Courier"><I>true</I></FONT>, the NORM receiver
is allowed to read received data from the stream for a given FEC code
block as soon as any content for a subsequent FEC code block is
received, regardless of the sender stream buffer size setting (see
<FONT FACE="Courier"><A HREF="#NormStreamOpen()">NormStreamOpen()</A></FONT>) or the receiver's
receive buffer size (see <FONT FACE="Courier"><A HREF="#NormStartReceiver()">NormStartReceiver()</A></FONT>).
This eliminates data delivery delay due to receiver buffering beyond
one FEC code block, but may result in compromise of potential stream
reliability where the sender may provide additional repair
transmissions in response to NACKs from other, non-silent receiver
participating in the <FONT FACE="Courier"><U>session</U></FONT>. The
default behavior (<FONT FACE="Courier"><U>lowDelay</U></FONT> =
<FONT FACE="Courier"><I>false</I></FONT>), causes the receiver to
buffer received FEC code blocks for as long as possible (within
buffer constraints as newer data arrives) before allowing the
application to read the data. This provides maximum reliability, but
buffer delays of potentially multiple FEC code blocks may be incurred
when packet loss occurs.</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">This function has no
return values.</P>
@ -3872,11 +4021,11 @@ NormNodeGetId(<A HREF="#NormNodeHandle">NormNodeHandle</A> nodeHandle)</FONT></F
<P CLASS="western" STYLE="font-style: normal">This function retrieves
the <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT> identifier for the remote
participant referenced by the given <FONT FACE="Courier"><U>nodeHandle</U></FONT>
value. The <A HREF="#NormNodeId">NormNodeId</A> is a 32-bit value used within the NORM
protocol to uniquely identify participants within a NORM session.
The participants identifiers are assigned by the application or
derived (by the NORM API code) from the host computers default IP
address.</P>
value. The <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT> is a 32-bit value
used within the NORM protocol to uniquely identify participants
within a NORM session. The participants identifiers are assigned by
the application or derived (by the NORM API code) from the host
computers default IP address.</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">This function returns
the <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT> value associated with the
@ -3888,24 +4037,55 @@ value will be <FONT FACE="Courier"><I>NORM_NODE_NONE</I></FONT>.</P>
<H4 CLASS="western">Synopsis</H4>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>#include
&lt;normApi.h&gt;</FONT></FONT></P>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2><A HREF="#NormNodeId">NormNodeId</A>
NormNodeGetAddress(<A HREF="#NormNodeHandle">NormNodeHandle</A> nodeHandle)</FONT></FONT></P>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>bool
NormNodeGetAddress(<A HREF="#NormNodeHandle">NormNodeHandle</A>
nodeHandle,<BR>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;char*&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;addrBuffer,<BR>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;unsigned
int*&nbsp;&nbsp;&nbsp;bufferLen,<BR>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;unsigned
short*&nbsp;port = (unsigned short*)0);</FONT></FONT></P>
<H4 CLASS="western">Description</H4>
<P CLASS="western" STYLE="font-style: normal">This function retrieves
the <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT> identifier for the remote
participant referenced by the given <FONT FACE="Courier"><U>nodeHandle</U></FONT>
value. The <A HREF="#NormNodeId">NormNodeId</A> is a 32-bit value used within the NORM
protocol to uniquely identify participants within a NORM session.
The participants identifiers are assigned by the application or
derived (by the NORM API code) from the host computers default IP
address.</P>
the current network source address detected for packets received from
remote NORM sender referenced by the <FONT FACE="Courier"><U>nodeHandle</U></FONT>
parameter. The <FONT FACE="Courier"><U>addrBuffer</U></FONT> must be
a pointer to storage of <FONT FACE="Courier"><U>bufferLen</U></FONT>
bytes in length in which the referenced sender node's address will be
returned. Optionally, the remote sender source port number (see
<FONT FACE="Courier"><A HREF="#NormSetTxPort()">NormSetTxPort()</A></FONT>) is also returned if the
optional <FONT FACE="Courier"><U>port</U></FONT> pointer to storage
parameter is provided in the call. Note that in the case of Network
Address Translation (NAT) or other firewall activities, the source
address detected by the NORM receiver may not be the original address
of the original NORM sender.</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">A value of <FONT FACE="Courier"><I>true</I></FONT>
is returned upon success and <FONT FACE="Courier"><I>false</I></FONT>
upon failure. An invalid <FONT FACE="Courier"><U>nodeHandle</U></FONT>
parameter value would lead to failure.</P>
<H3 CLASS="functionheading-western"><A NAME="NormNodeGetGrtt()">NormNodeGetGrtt()</A></H3>
<H4 CLASS="western">Synopsis</H4>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>#include
&lt;normApi.h&gt;</FONT></FONT></P>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>double
NormNodeGetId(<A HREF="#NormNodeHandle">NormNodeHandle</A> nodeHandle)</FONT></FONT></P>
<H4 CLASS="western">Description</H4>
<P CLASS="western" STYLE="font-style: normal">This function retrieves
the advertised estimate of group round-trip timing (GRTT) for the
remote sender referenced by the given <FONT FACE="Courier"><U>nodeHandle</U></FONT>
value. Newly-starting senders that have been participating as a
receiver within a group may wish to use this function to provide a
more accurate startup estimate of GRTT (see <FONT FACE="Courier"><A HREF="#NormSetGrttEstimate()">NormSetGrttEstimate()</A></FONT>)
prior to a call to <FONT FACE="Courier"><A HREF="#NormStartSender()">NormStartSender()</A></FONT>.
Applications may use this information for other purpose as well.
Note that the <FONT FACE="Courier"><I>NORM_GRTT_UPDATED</I></FONT>
event is posted (see <FONT FACE="Courier"><A HREF="#NormGetNextEvent()">NormGetNextEvent()</A></FONT>)
by the NORM protocol engine to indicate when changes in the local
sender or remote senders' GRTT estimate occurs.</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">This function returns
the <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT> value associated with the
specified <FONT FACE="Courier"><U>nodeHandle</U></FONT>. In the case
<FONT FACE="Courier"><U>nodeHandle</U></FONT> is equal to
<FONT FACE="Courier"><I>NORM_NODE_INVALID</I></FONT>, the return
value will be <FONT FACE="Courier"><I>NORM_NODE_NONE</I></FONT>.</P>
the remote sender's advertised GRTT estimate in units of seconds. A
value of <FONT FACE="Courier"><I>-1.0</I></FONT> is returned upon
failure. An invalid <FONT FACE="Courier"><U>nodeHandle</U></FONT>
parameter value will lead to such failure.</P>
<H3 CLASS="functionheading-western"><A NAME="NormNodeRetain()">NormNodeRetain()</A></H3>
<H4 CLASS="western">Synopsis</H4>
<P STYLE="font-style: normal; page-break-after: avoid"><FONT FACE="Courier"><FONT SIZE=2>#include
@ -3923,31 +4103,31 @@ associated state and thus invalidate the <FONT FACE="Courier"><A HREF="#NormNode
call to <FONT FACE="Courier"><A HREF="#NormNodeRelease()">NormNodeRelease()</A></FONT> when finished
with the node information to avoid a memory leak condition.
<FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT> values (unless retained)
are valid from the time of a <FONT FACE="Courier"><I>NORM_REMOTE_SERVER_NEW</I></FONT>
notification until a complimentary <FONT FACE="Courier"><I>NORM_REMOTE_SERVER_PURGED</I></FONT>
are valid from the time of a <FONT FACE="Courier"><I>NORM_REMOTE_SENDER_NEW</I></FONT>
notification until a complimentary <FONT FACE="Courier"><I>NORM_REMOTE_SENDER_PURGED</I></FONT>
notification. During that interval, the application will receive
<FONT FACE="Courier"><I>NORM_REMOTE_SERVER_ACTIVE</I></FONT> and
<FONT FACE="Courier"><I>NORM_REMOTE_SERVER_INACTIVE</I></FONT>
notifications according to the server's message transmission activity
<FONT FACE="Courier"><I>NORM_REMOTE_SENDER_ACTIVE</I></FONT> and
<FONT FACE="Courier"><I>NORM_REMOTE_SENDER_INACTIVE</I></FONT>
notifications according to the sender's message transmission activity
within the session.</P>
<P CLASS="western" STYLE="font-style: normal">It is important to note
that, if the NORM protocol engine posts a <FONT FACE="Courier"><I>NORM_REMOTE_SERVER_PURGED</I></FONT>
that, if the NORM protocol engine posts a <FONT FACE="Courier"><I>NORM_REMOTE_SENDER_PURGED</I></FONT>
notification for a given <FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT>,
the NORM protocol engine could possibly, subsequently establish a
new, different <FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT> value for
the same remote server (i.e., one of equivalent <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT>)
the same remote sender (i.e., one of equivalent <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT>)
if it again becomes active in the session. A new <FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT>
may likely be established even if the application has retained the
previous <FONT FACE="Courier"><A HREF="#NormNodeHandle">NormNodeHandle</A></FONT> value.
Therefore, to the application, it might appear that two different
servers with the same <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT> are
senders with the same <FONT FACE="Courier"><A HREF="#NormNodeId">NormNodeId</A></FONT> are
participating if these notifications are not carefully monitored.
This behavior is contingent upon how the application has configured
the NORM protocol engine to manage resources when there is potential
for a large number of remote servers within a session (related APIs
for a large number of remote senders within a session (related APIs
are TBD). For example, the application may wish to control which
specific remote servers for which it keeps state (or limit the memory
resources used for remote servers state, etc) and the NORM API may be
specific remote senders for which it keeps state (or limit the memory
resources used for remote sender state, etc) and the NORM API may be
extended in the future to control this behavior.</P>
<H4 CLASS="western">Return Values</H4>
<P CLASS="western" STYLE="font-style: normal">This function has no

BIN
NormDeveloperGuide.pdf
View File

Binary file not shown.

8
TODO.TXT
View File

@ -37,6 +37,14 @@
12) Can we eliminate rapid clr norm-cc probing when data is
not being actively sent? (COMPLETED)
13) Fix ns-2 issue when Agent/NORM is attached to ns-2 node ZERO and
results in a NormNode attempting to be assigned NORM_NODE_NONE as
a node id.
14) IMPORANT: double check ACKing in reponse to watermark polling and
relation to cc feedback suppression, etc ... watermark acks should
be unicast and not effect cc feedback?

35
VERSION.TXT
View File

@ -1,5 +1,40 @@
NORM Version History
Version 1.3b9
=============
- Fixed ProtoSocket WINVER stuff and NormApplicationMsg::SetContent()
(thanks to Stefan Titze for these)
- Fixed issue in NormSession::OnRepairTimeout() where a debug
message erroneously dereferenced a NormObject pointer as
NormFileObject (led to run-time issue in optimized code build)
(thanks to Charlie Davis once again)
Version 1.3b8
=============
- Fixed implementation of NormQuantizeGroupSize() in "normMessage.h"
(Thanks to David Lucas for finding the bug)
- Fixed some issues with NORM/Protolib ns-2 support
- Added API NormGetGrtt() and NormNodeGetGrtt() calls
to support some application needs.
- Modified transmission of FEC parity segments such that their
length is only as big as needed to support the maximum
sent segment size for the associated block (this allows more
latitude for stream apps in setting their sender segment
size and flushing).
- Added NORM_CC_ACTIVE and NORM_CC_INACTIVE notifications to indicate
whether receivers are actively providing congestion control feedback
(Thanks to Charlie Davis for the great suggestion)
- Fixed some floating-point exceptions that could occur in receiver
congestion control tracking under extreme circumstances. This included
enhancement of receiver receive rate tracking.
(Thanks to Ian Downard for finding these issues)
- Changed naming of "SERVER" notifications in API to "SENDER"
notifications to be consistent with the rest of API.
- Debug statistics now indicate correct nack/suppressed counts
for silent receiver operation (i.e. none)
- Refined activity timeout of remote sender
(still need to give the app more control over resource allocation)
Version 1.3b7
=============
- Deprecated NORM stream payload flags in favor of "payload_msg_start"

196
common/normApi.cpp
View File

@ -40,7 +40,7 @@ class NormInstance : public NormController
void Notify(NormController::Event event,
class NormSessionMgr* sessionMgr,
class NormSession* session,
class NormServerNode* server,
class NormServerNode* sender,
class NormObject* object);
bool Startup(bool priorityBoost = false);
@ -140,7 +140,7 @@ class NormInstance : public NormController
return n;
}
Notification* GetHead() {return head;}
void SetHead(Notification* n) {head = n;}
void SetTail(Notification* n) {tail = n;}
private:
Notification* head;
Notification* tail;
@ -245,7 +245,7 @@ bool NormInstance::SetCacheDirectory(const char* cachePath)
void NormInstance::Notify(NormController::Event event,
class NormSessionMgr* sessionMgr,
class NormSession* session,
class NormServerNode* server,
class NormServerNode* sender,
class NormObject* object)
{
// (TBD) set a limit on how many pending notifications
@ -379,20 +379,20 @@ void NormInstance::Notify(NormController::Event event,
break;
} // end case RX_OBJECT_NEW
case RX_OBJECT_COMPLETED:
case TX_OBJECT_PURGED:
case RX_OBJECT_ABORTED:
object->Retain();
break;
default:
break;
} // end switch(event)
// "Retain" any valid "object" or "sender" handles for API access
if (NORM_OBJECT_INVALID != object)
((NormObject*)object)->Retain();
else if (NORM_NODE_INVALID != sender)
((NormServerNode*)sender)->Retain();
bool doNotify = notify_queue.IsEmpty();
n->event.type = (NormEventType)event;
n->event.session = session;
n->event.sender = server;
n->event.sender = sender;
n->event.object = object;
notify_queue.Append(n);
@ -421,25 +421,30 @@ void NormInstance::Notify(NormController::Event event,
void NormInstance::PurgeObjectNotifications(NormObjectHandle objectHandle)
{
if (NORM_OBJECT_INVALID == objectHandle) return;
Notification* prev = NULL;
Notification* next = notify_queue.GetHead();
while (next)
{
// TBD - return these to the pool?
if (objectHandle == next->event.object)
{
switch (next->event.type)
{
case RX_OBJECT_COMPLETED:
case TX_OBJECT_PURGED:
case RX_OBJECT_ABORTED:
((NormObject*)(next->event.object))->Release();
break;
default:
break;
}
next->event.type = NORM_EVENT_INVALID;
// "Release" the previously-retained object handle
((NormObject*)objectHandle)->Release();
// Remove this notification from queue and return to pool
Notification* current = next;
next = next->GetNext();
if (NULL != prev)
prev->Append(next);
else
notify_queue.RemoveHead();
if (NULL == next) notify_queue.SetTail(prev);
notify_pool.Append(current);
}
else
{
prev = next;
next = next->GetNext();
}
next = next->GetNext();
}
// (TBD) if we put these in the pool, should we check that the queue has been emptied
// and possible reset event/fd
@ -451,20 +456,11 @@ bool NormInstance::GetNextEvent(NormEvent* theEvent)
// First, do any garbage collection for "previous_notification"
if (NULL != previous_notification)
{
// "Release" any purged/completed/aborted objects
switch(previous_notification->event.type)
{
case NORM_RX_OBJECT_COMPLETED:
case NORM_TX_OBJECT_PURGED:
case NORM_RX_OBJECT_ABORTED:
{
NormObject* obj = (NormObject*)(previous_notification->event.object);
obj->Release();
break;
}
default:
break;
}
// "Release" any previously-retained object or node handle
if (NORM_OBJECT_INVALID != previous_notification->event.object)
((NormObject*)(previous_notification->event.object))->Release();
else if (NORM_NODE_INVALID != previous_notification->event.object)
((NormServerNode*)(previous_notification->event.sender))->Release();
notify_pool.Append(previous_notification);
previous_notification = NULL;
}
@ -487,16 +483,17 @@ bool NormInstance::GetNextEvent(NormEvent* theEvent)
switch (n->event.type)
{
case NORM_EVENT_INVALID:
continue;
if (!notify_queue.IsEmpty())
{
// Discard this invalid event and get next one
notify_pool.Append(n);
continue;
}
break;
case NORM_RX_OBJECT_UPDATED:
// reset update event notification
((NormObject*)n->event.object)->SetNotifyOnUpdate(true);
break;
case NORM_TX_QUEUE_VACANCY:
//if (NULL != n->event.object)
// static_cast<NormStreamObject*>((NormObject*)n->event.object)->ResetPostedTxQueueVacancy();
// (TBD) support session-level queue vacancy notify
break;
default:
break;
}
@ -600,19 +597,11 @@ void NormInstance::Shutdown()
// Garbage collect our "previous_notification"
if (NULL != previous_notification)
{
switch(previous_notification->event.type)
{
case NORM_RX_OBJECT_COMPLETED:
case NORM_TX_OBJECT_PURGED:
case NORM_RX_OBJECT_ABORTED:
{
NormObject* obj = (NormObject*)(previous_notification->event.object);
obj->Release();
break;
}
default:
break;
}
// Release any previously-retained object or node handles
if (NORM_OBJECT_INVALID != previous_notification->event.object)
((NormObject*)(previous_notification->event.object))->Release();
else if (NORM_NODE_INVALID != previous_notification->event.object)
((NormServerNode*)(previous_notification->event.sender))->Release();
notify_pool.Append(previous_notification);
previous_notification = NULL;
}
@ -635,15 +624,13 @@ void NormInstance::Shutdown()
}
break;
}
// Garbage collect retained objects
case NORM_RX_OBJECT_COMPLETED:
case NORM_TX_OBJECT_PURGED:
case NORM_RX_OBJECT_ABORTED:
((NormObject*)n->event.object)->Release();
break;
default:
break;
break;
}
if (NORM_OBJECT_INVALID != n->event.object)
((NormObject*)(n->event.object))->Release();
else if (NORM_NODE_INVALID != n->event.object)
((NormServerNode*)(n->event.sender))->Release();
delete n;
}
notify_pool.Destroy();
@ -1106,6 +1093,23 @@ void NormSetGrttEstimate(NormSessionHandle sessionHandle,
}
} // end NormSetGrttEstimate()
NORM_API_LINKAGE
double NormGetGrttEstimate(NormSessionHandle sessionHandle)
{
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormSession* session = (NormSession*)sessionHandle;
if (session) session->ServerGrtt();
instance->dispatcher.ResumeThread();
}
if (NORM_SESSION_INVALID != sessionHandle)
return (((NormSession*)sessionHandle)->ServerGrtt());
else
return -1.0;
} // end NormGetGrttEstimate()
NORM_API_LINKAGE
void NormSetGrttMax(NormSessionHandle sessionHandle,
double grttMax)
@ -1186,8 +1190,9 @@ NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle,
NormSession* session = (NormSession*)sessionHandle;
if (session)
{
NormObject* obj = session->QueueTxFile(fileName, infoPtr, infoLen);
if (obj) objectHandle = (NormObjectHandle)obj;
NormObject* obj =
static_cast<NormObject*>(session->QueueTxFile(fileName, infoPtr, infoLen));
if (obj) objectHandle = (NormObjectHandle)(obj);
}
instance->dispatcher.ResumeThread();
}
@ -1208,7 +1213,8 @@ NormObjectHandle NormDataEnqueue(NormSessionHandle sessionHandle,
NormSession* session = (NormSession*)sessionHandle;
if (session)
{
NormObject* obj = session->QueueTxData(dataPtr, dataLen, infoPtr, infoLen);
NormObject* obj =
static_cast<NormObject*>(session->QueueTxData(dataPtr, dataLen, infoPtr, infoLen));
if (obj) objectHandle = (NormObjectHandle)obj;
}
instance->dispatcher.ResumeThread();
@ -1216,6 +1222,25 @@ NormObjectHandle NormDataEnqueue(NormSessionHandle sessionHandle,
return objectHandle;
} // end NormDataEnqueue()
NORM_API_LINKAGE
bool NormRequeueObject(NormSessionHandle sessionHandle, NormObjectHandle objectHandle)
{
bool result = false;
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormSession* session = (NormSession*)sessionHandle;
if (session)
{
if (NORM_OBJECT_INVALID != objectHandle)
result = session->RequeueTxObject((NormObject*)objectHandle);
}
instance->dispatcher.ResumeThread();
}
return result;
} // end NormRequeueObject()
NORM_API_LINKAGE
NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle,
unsigned long bufferSize)
@ -1476,10 +1501,15 @@ bool NormSetRxSocketBuffer(NormSessionHandle sessionHandle,
NORM_API_LINKAGE
void NormSetSilentReceiver(NormSessionHandle sessionHandle,
bool silent)
bool silent,
bool lowDelay)
{
NormSession* session = (NormSession*)sessionHandle;
if (session) session->ClientSetSilent(silent);
if (session)
{
session->ClientSetSilent(silent);
session->ReceiverSetLowDelay(lowDelay);
}
} // end NormSetSilentReceiver()
NORM_API_LINKAGE
@ -1494,7 +1524,8 @@ NORM_API_LINKAGE
void NormNodeSetUnicastNack(NormNodeHandle nodeHandle,
bool unicastNacks)
{
NormServerNode* node = (NormServerNode*)nodeHandle;
NormServerNode* node = (NormServerNode*)nodeHandle;
if (node) node->SetUnicastNacks(unicastNacks);
} // end NormNodeSetUnicastNack()
@ -1679,8 +1710,7 @@ void NormObjectRetain(NormObjectHandle objectHandle)
NormInstance* instance = NormInstance::GetInstanceFromObject(objectHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormObject* obj = (NormObject*)objectHandle;
obj->Retain();
((NormObject*)objectHandle)->Retain();
instance->dispatcher.ResumeThread();
}
}
@ -1689,14 +1719,14 @@ void NormObjectRetain(NormObjectHandle objectHandle)
NORM_API_LINKAGE
void NormObjectRelease(NormObjectHandle objectHandle)
{
// (TBD) we could maintain separate "app_retain" and "interval_retain"
// counts to prevent bad apps from messing up NORM interval code ???
if (NORM_OBJECT_INVALID != objectHandle)
{
NormInstance* instance = NormInstance::GetInstanceFromObject(objectHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormObject* obj = (NormObject*)objectHandle;
obj->Release();
instance->PurgeObjectNotifications(objectHandle);
((NormObject*)objectHandle)->Release();
instance->dispatcher.ResumeThread();
}
}
@ -1805,6 +1835,16 @@ bool NormNodeGetAddress(NormNodeHandle nodeHandle,
return result;
} // end NormNodeGetId()
NORM_API_LINKAGE
double NormNodeGetGrtt(NormNodeHandle nodeHandle)
{
NormServerNode* node = (NormServerNode*)nodeHandle;
if (NULL != node)
return node->GetGrttEstimate();
else
return -1.0;
} // end NormNodeGetGrtt()
NORM_API_LINKAGE
void NormNodeRetain(NormNodeHandle nodeHandle)
{
@ -1813,8 +1853,7 @@ void NormNodeRetain(NormNodeHandle nodeHandle)
NormInstance* instance = NormInstance::GetInstanceFromNode(nodeHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormNode* node = (NormNode*)nodeHandle;
node->Retain();
((NormServerNode*)nodeHandle)->Retain();
instance->dispatcher.ResumeThread();
}
}
@ -1828,8 +1867,7 @@ void NormNodeRelease(NormNodeHandle nodeHandle)
NormInstance* instance = NormInstance::GetInstanceFromNode(nodeHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormNode* node = (NormNode*)nodeHandle;
node->Release();
((NormServerNode*)nodeHandle)->Release();
instance->dispatcher.ResumeThread();
}
}
@ -1848,6 +1886,8 @@ void NormNodeRelease(NormNodeHandle nodeHandle)
// NORM thread ... an approach to this would be enable
// the app to install an event handler callback and dispatch
// events with a "NormDispatchEvents()" call ...
// (Update: I've improved the way the NORM thread suspend/resume
// is done so that this current approach is not very costly)
//
NORM_API_LINKAGE

29
common/normApi.h
View File

@ -124,16 +124,19 @@ enum NormEventType
NORM_TX_WATERMARK_COMPLETED,
NORM_TX_OBJECT_SENT,
NORM_TX_OBJECT_PURGED,
NORM_LOCAL_SERVER_CLOSED,
NORM_REMOTE_SERVER_NEW,
NORM_REMOTE_SERVER_ACTIVE,
NORM_REMOTE_SERVER_INACTIVE,
NORM_REMOTE_SERVER_PURGED,
NORM_LOCAL_SENDER_CLOSED,
NORM_REMOTE_SENDER_NEW,
NORM_REMOTE_SENDER_ACTIVE,
NORM_REMOTE_SENDER_INACTIVE,
NORM_REMOTE_SENDER_PURGED,
NORM_RX_OBJECT_NEW,
NORM_RX_OBJECT_INFO,
NORM_RX_OBJECT_UPDATED,
NORM_RX_OBJECT_COMPLETED,
NORM_RX_OBJECT_ABORTED
NORM_RX_OBJECT_ABORTED,
NORM_GRTT_UPDATED,
NORM_CC_ACTIVE,
NORM_CC_INACTIVE
};
typedef struct
@ -159,6 +162,7 @@ NORM_API_LINKAGE
bool NormRestartInstance(NormInstanceHandle instanceHandle);
// This MUST be set to enable NORM_OBJECT_FILE reception!
// (otherwise received files are ignored)
NORM_API_LINKAGE
bool NormSetCacheDirectory(NormInstanceHandle instanceHandle,
const char* cachePath);
@ -286,6 +290,9 @@ NORM_API_LINKAGE
void NormSetGrttEstimate(NormSessionHandle sessionHandle,
double grttEstimate);
NORM_API_LINKAGE
double NormGetGrttEstimate(NormSessionHandle sessionHandle);
NORM_API_LINKAGE
void NormSetGrttMax(NormSessionHandle sessionHandle,
double grttMax);
@ -320,7 +327,9 @@ NormObjectHandle NormDataEnqueue(NormSessionHandle sessionHandle,
const char* infoPtr = (const char*)0,
unsigned int infoLen = 0);
NORM_API_LINKAGE
bool NormRequeueObject(NormSessionHandle sessionHandle, NormObjectHandle objectHandle);
NORM_API_LINKAGE
NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle,
unsigned long bufferSize);
@ -383,7 +392,8 @@ bool NormSetRxSocketBuffer(NormSessionHandle sessionHandle,
NORM_API_LINKAGE
void NormSetSilentReceiver(NormSessionHandle sessionHandle,
bool silent);
bool silent,
bool lowDelay = false);
NORM_API_LINKAGE
void NormSetDefaultUnicastNack(NormSessionHandle sessionHandle,
@ -488,6 +498,9 @@ bool NormNodeGetAddress(NormNodeHandle nodeHandle,
unsigned int* bufferLen,
unsigned short* port = (unsigned short*)0);
NORM_API_LINKAGE
double NormNodeGetGrtt(NormNodeHandle nodeHandle);
NORM_API_LINKAGE
void NormNodeRetain(NormNodeHandle nodeHandle);

19
common/normApp.cpp
View File

@ -87,6 +87,7 @@ class NormApp : public NormController, public ProtoApp
// NormSession common parameters
char* address; // session address
UINT16 port; // session port number
UINT16 tx_port;
UINT8 ttl;
bool loopback;
char* interface_name; // for multi-home hosts
@ -196,7 +197,8 @@ const char* const NormApp::cmd_list[] =
"+trace", // message tracing on
"+txloss", // tx packet loss percent (for testing)
"+rxloss", // rx packet loss percent (for testing)
"+address", // session destination address
"+address", // session destination address/port
"+txport", // use specific transmission source port number
"+ttl", // multicast hop count scope
"+loopback", // "off" or "on" to recv our own packets
"+interface", // multicast interface name to use
@ -247,6 +249,7 @@ void NormApp::ShowHelp()
" +txloss, // tx packet loss percent (for testing)\n"
" +rxloss, // rx packet loss percent (for testing)\n"
" +address, // session destination address\n"
" +txport, // use a specific transmission source port number\n"
" +ttl, // multicast hop count scope\n"
" +loopback, // 'on' or 'off' to recv our own packets (default = 'off')\n"
" +interface, // multicast interface name to use\n"
@ -448,6 +451,17 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
}
port = portNum;
}
else if (!strncmp("txport", cmd, len))
{
int txPort = atoi(val);
if ((txPort < 0) || (txPort > 65535))
{
DMSG(0, "NormApp::OnCommand(txport) invalid port number!\n");
return false;
}
tx_port = (UINT16)txPort;
if (session) session->SetTxPort(txPort);
}
else if (!strncmp("ttl", cmd, len))
{
int ttlTemp = atoi(val);
@ -1557,13 +1571,14 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
if (session)
{
// Common session parameters
session->SetTxPort(tx_port);
session->SetTxRate(tx_rate);
session->SetTrace(tracing);
session->SetTxLoss(tx_loss);
session->SetRxLoss(rx_loss);
session->SetTTL(ttl);
session->SetLoopback(loopback);
if (input || !tx_file_list.IsEmpty())
{
NormObjectId baseId = (unsigned short)(rand() * (65535.0/ (double)RAND_MAX));

15
common/normFile.cpp
View File

@ -413,8 +413,8 @@ NormFile::Offset NormFile::GetSize() const
return ((Offset)fileSize);
#else
#ifdef WIN32
struct __stat64 info; // instead of "struct _stat"
int result = _fstat64(fd, &info); // instead of "_fstat()"
struct _stati64 info; // instead of "struct _stat"
int result = _fstati64(fd, &info); // instead of "_fstat()"
#else
struct stat info;
int result = fstat(fd, &info);
@ -827,8 +827,8 @@ NormFile::Offset NormFile::GetSize(const char* path)
}
#else
#ifdef WIN32
struct __stat64 info; // instead of "struct _stat"
int result = _stat64(path, &info); // instead of "_stat()"
struct _stati64 info; // instead of "struct _stat"
int result = _stati64(path, &info); // instead of "_stat()"
#else
struct stat info;
int result = stat(path, &info);
@ -879,8 +879,8 @@ time_t NormFile::GetUpdateTime(const char* path)
}
#else
#ifdef WIN32
struct __stat64 info; // instead of "struct _stat"
int result = _stat64(path, &info); // instead of "_stat()"
struct _stati64 info; // instead of "struct _stat"
int result = _stati64(path, &info); // instead of "_stat()"
#else
struct stat info;
int result = stat(path, &info);
@ -893,7 +893,8 @@ time_t NormFile::GetUpdateTime(const char* path)
{
#ifdef WIN32
// Hack because Win2K and Win98 seem to work differently
__time64_t updateTime = MAX(info.st_ctime, info.st_atime);
//__time64_t updateTime = MAX(info.st_ctime, info.st_atime);
time_t updateTime = MAX(info.st_ctime, info.st_atime);
updateTime = MAX(updateTime, info.st_mtime);
return ((time_t)updateTime);
#else

29
common/normMessage.h
View File

@ -71,7 +71,7 @@ inline unsigned char NormQuantizeGroupSize(double gsize)
int mantissa = (int)((gsize/pow(10.0, (double)ebits)) + 0.5);
// round up quantized group size
unsigned char mbit = (mantissa > 1) ? ((mantissa > 5) ? 0x00 : 0x08) : 0x00;
ebits = (mantissa > 5) ? ebits : ebits+1;
ebits = (mantissa > 5) ? ebits : ebits-1;
mbit = (ebits > 0x07) ? 0x08 : mbit;
ebits = (ebits > 0x07) ? 0x07 : ebits;
return (mbit | ebits);
@ -91,6 +91,7 @@ inline double NormUnquantizeLoss(unsigned short lossQuantized)
inline unsigned short NormQuantizeRate(double rate)
{
if (rate <= 0.0) return 0x01; // rate = 0.0
unsigned char exponent = (unsigned char)log10(rate);
unsigned short mantissa = (unsigned short)((256.0/10.0) * rate / pow(10.0, (double)exponent));
return ((mantissa << 8) | exponent);
@ -323,9 +324,12 @@ class NormMsg
// Message processing routines
bool InitFromBuffer(UINT16 msgLength);
UINT8 GetVersion() const {return ((UINT8*)buffer)[VERSION_OFFSET];}
NormMsg::Type GetType() const {return (Type)(((UINT8*)buffer)[TYPE_OFFSET] & 0x0f);}
UINT16 GetHeaderLength() {return ((UINT8*)buffer)[HDR_LEN_OFFSET] << 2;}
UINT8 GetVersion() const
{return (((UINT8*)buffer)[VERSION_OFFSET] >> 4);}
NormMsg::Type GetType() const
{return (Type)(((UINT8*)buffer)[TYPE_OFFSET] & 0x0f);}
UINT16 GetHeaderLength()
{return ((UINT8*)buffer)[HDR_LEN_OFFSET] << 2;}
UINT16 GetSequence() const
{
return (ntohs((((UINT16*)buffer)[SEQUENCE_OFFSET])));
@ -1363,6 +1367,7 @@ class NormCmdApplicationMsg : public NormCmdMsg
{
UINT16 len = MIN(contentLen, segmentSize);
memcpy(((char*)buffer)+header_length, content, len);
length = header_length + len;
return (contentLen <= segmentSize);
}
@ -1391,7 +1396,7 @@ class NormNackMsg : public NormMsg
// Message building
void SetServerId(NormNodeId serverId)
{
buffer[SERVER_ID_OFFSET] = htonl(serverId);
buffer[SENDER_ID_OFFSET] = htonl(serverId);
}
void SetInstanceId(UINT16 instanceId)
{
@ -1419,7 +1424,7 @@ class NormNackMsg : public NormMsg
// Message processing
NormNodeId GetServerId() const
{
return (ntohl(buffer[SERVER_ID_OFFSET]));
return (ntohl(buffer[SENDER_ID_OFFSET]));
}
UINT16 GetInstanceId() const
{
@ -1445,8 +1450,8 @@ class NormNackMsg : public NormMsg
private:
enum
{
SERVER_ID_OFFSET = MSG_OFFSET/4,
INSTANCE_ID_OFFSET = ((SERVER_ID_OFFSET*4)+4)/2,
SENDER_ID_OFFSET = MSG_OFFSET/4,
INSTANCE_ID_OFFSET = ((SENDER_ID_OFFSET*4)+4)/2,
RESERVED_OFFSET = ((INSTANCE_ID_OFFSET*2)+2)/2,
GRTT_RESPONSE_SEC_OFFSET = ((RESERVED_OFFSET*2)+2)/4,
GRTT_RESPONSE_USEC_OFFSET = ((GRTT_RESPONSE_SEC_OFFSET*4)+4)/4,
@ -1467,7 +1472,7 @@ class NormAckMsg : public NormMsg
}
void SetServerId(NormNodeId serverId)
{
buffer[SERVER_ID_OFFSET] = htonl(serverId);
buffer[SENDER_ID_OFFSET] = htonl(serverId);
}
void SetInstanceId(UINT16 instanceId)
{
@ -1491,7 +1496,7 @@ class NormAckMsg : public NormMsg
// Message processing
NormNodeId GetServerId() const
{
return (ntohl(buffer[SERVER_ID_OFFSET]));
return (ntohl(buffer[SENDER_ID_OFFSET]));
}
UINT16 GetInstanceId() const
{
@ -1510,8 +1515,8 @@ class NormAckMsg : public NormMsg
protected:
enum
{
SERVER_ID_OFFSET = MSG_OFFSET/4,
INSTANCE_ID_OFFSET = ((SERVER_ID_OFFSET*4)+4)/2,
SENDER_ID_OFFSET = MSG_OFFSET/4,
INSTANCE_ID_OFFSET = ((SENDER_ID_OFFSET*4)+4)/2,
ACK_TYPE_OFFSET = (INSTANCE_ID_OFFSET*2)+2,
ACK_ID_OFFSET = ACK_TYPE_OFFSET + 1,
GRTT_RESPONSE_SEC_OFFSET = (ACK_ID_OFFSET + 1)/4,

189
common/normNode.cpp
View File

@ -1,4 +1,3 @@
#include "normNode.h"
#include "normSession.h"
@ -42,6 +41,9 @@ NormCCNode::~NormCCNode()
}
const double NormServerNode::DEFAULT_NOMINAL_INTERVAL = 2*NormSession::DEFAULT_GRTT_ESTIMATE;
const double NormServerNode::ACTIVITY_INTERVAL_MIN = 1.0; // 1 second min activity timeout
NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
: NormNode(theSession, nodeId), instance_id(0), synchronized(false), sync_id(0),
max_pending_range(256), is_open(false), segment_size(0), ndata(0), nparity(0),
@ -65,7 +67,9 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
repair_timer.SetRepeat(1);
activity_timer.SetListener(this, &NormServerNode::OnActivityTimeout);
activity_timer.SetInterval(NormSession::DEFAULT_GRTT_ESTIMATE*NORM_ROBUST_FACTOR);
double activityInterval = NormSession::DEFAULT_GRTT_ESTIMATE*NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
activity_timer.SetRepeat(NORM_ROBUST_FACTOR);
cc_timer.SetListener(this, &NormServerNode::OnCCTimeout);
@ -93,6 +97,8 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
prev_update_time.tv_sec = 0;
prev_update_time.tv_usec = 0;
recv_rate_quantized = NormQuantizeRate(recv_rate);
}
NormServerNode::~NormServerNode()
@ -167,17 +173,22 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize,
unsigned long segPerBlock =
(unsigned long) ((bufferFactor * (double)numData) +
((1.0 - bufferFactor) * (double)numParity) + 0.5);
if (segPerBlock > numData) segPerBlock = numData;
// If there's no parity, no segment buffering for decoding is required at all!
// (Thus, the full rxbuffer space can be used for block state)
if (0 == numParity) segPerBlock = 0;
unsigned long blockSegmentSpace = segPerBlock * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength());
unsigned long blockSpace = blockStateSpace+blockSegmentSpace;
unsigned long blockSpace = blockStateSpace + blockSegmentSpace;
unsigned long numBlocks = bufferSpace / blockSpace;
// Round numBlocks upward
if (bufferSpace > (numBlocks*blockSpace)) numBlocks++;
// Always have at least 2 blocks in the pool
if (numBlocks < 2) numBlocks = 2;
unsigned long numSegments = numBlocks * segPerBlock;
if (!block_pool.Init(numBlocks, blockSize))
{
DMSG(0, "NormServerNode::AllocateBuffers() block_pool init error\n");
@ -299,8 +310,17 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime,
grtt_estimate = NormUnquantizeRtt(grttQuantized);
DMSG(4, "NormServerNode::HandleCommand() node>%lu server>%lu new grtt: %lf sec\n",
LocalNodeId(), GetId(), grtt_estimate);
activity_timer.SetInterval(grtt_estimate*NORM_ROBUST_FACTOR);
// Possible change in activity timeout
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double activityInterval =
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
if (activity_timer.IsActive()) activity_timer.Reschedule();
session.Notify(NormController::GRTT_UPDATED, this, (NormObject*)NULL);
}
UINT8 gsizeQuantized = cmd.GetGroupSize();
if (gsizeQuantized != gsize_quantized)
@ -396,9 +416,9 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime,
backoffFactor = MAX(backoffFactor, 4.0);
maxBackoff = grtt_estimate*backoffFactor;
}
double backoffTime = (maxBackoff > 0.0) ?
ExponentialRand(maxBackoff, gsize_estimate) :
0.0;
double backoffTime =
(maxBackoff > 0.0) ?
ExponentialRand(maxBackoff, gsize_estimate) : 0.0;
// Bias backoff timeout based on our rate
double r;
if (slow_start)
@ -567,11 +587,14 @@ void NormServerNode::HandleCCFeedback(UINT8 ccFlags, double ccRate)
}
if (suppressed)
{
if (cc_timer.IsActive()) cc_timer.Deactivate();
// This sets a holdoff timeout for cc feedback when suppressed
double backoffFactor = backoff_factor;
backoffFactor = MAX(backoffFactor, 4.0); // always use at least 4 for cc purposes
cc_timer.SetInterval(grtt_estimate*backoffFactor); // (TBD) ??? xxx
session.ActivateTimer(cc_timer);
cc_timer.SetInterval(grtt_estimate*backoffFactor);
if (cc_timer.IsActive())
cc_timer.Reschedule();
else
session.ActivateTimer(cc_timer);
cc_timer.DecrementRepeatCount();
}
}
@ -873,8 +896,20 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
grtt_estimate = NormUnquantizeRtt(grttQuantized);
DMSG(4, "NormServerNode::HandleObjectMessage() node>%lu server>%lu new grtt: %lf sec\n",
LocalNodeId(), GetId(), grtt_estimate);
activity_timer.SetInterval(grtt_estimate*NORM_ROBUST_FACTOR);
// Scale and reschedule pertinent timeouts repair_timer
// Possible change in sender activity timeout
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double activityInterval =
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
if (activity_timer.IsActive()) activity_timer.Reschedule();
session.Notify(NormController::GRTT_UPDATED, this, (NormObject*)NULL);
}
UINT8 gsizeQuantized = msg.GetGroupSize();
if (gsizeQuantized != gsize_quantized)
@ -1489,14 +1524,14 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
}
DMSG(6, "NormServerNode::OnRepairTimeout() node>%lu sending NACK rate:%lf kbps (rtt:%lf loss:%lf s:%hu) slow_start:%d\n",
LocalNodeId(), 8.0e-03*NormUnquantizeRate(ext.GetCCRate()),
rtt_estimate, ccLoss, nominal_packet_size, slow_start);
rtt_estimate, ccLoss, (UINT16)nominal_packet_size, slow_start);
ext.SetCCSequence(cc_sequence);
// Cancel potential pending NORM_ACK(RTT)
if (cc_timer.IsActive())
{
cc_timer.Deactivate();
// Set holdoff timeout to refrain from send too much cc feedback
cc_timer.SetInterval(grtt_estimate*backoff_factor);
session.ActivateTimer(cc_timer);
cc_timer.Reschedule();
cc_timer.DecrementRepeatCount();
}
} // end if (cc_enable)
@ -1632,10 +1667,13 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
if (nackAppended)
{
ASSERT(nack->GetRepairContentLength() > 0);
//session.QueueMessage(nack);
session.SendMessage(*nack);
if (!session.ClientIsSilent())
{
session.SendMessage(*nack);
nack_count++;
}
session.ReturnMessageToPool(nack);
nack_count++;
}
else
{
@ -1649,9 +1687,12 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
}
else
{
suppress_count++;
DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu NACK SUPPRESSED ...\n",
LocalNodeId());
if (!session.ClientIsSilent())
{
suppress_count++;
DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu NACK SUPPRESSED ...\n",
LocalNodeId());
}
} // end if/else(repairPending)
// BACKOFF related code
double holdoffInterval =
@ -1693,51 +1734,83 @@ void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned
double measurementInterval = rtt_confirmed ? rtt_estimate : grtt_estimate;
// Here, we put a NORM_TICK_MIN sec lower bound on our measurementInterval for the
// recv_rate because of the typical limited granularity of our system clock
// (Note this does limit our ramp up of data rate during slow start)
if (measurementInterval < NORM_TICK_MIN) measurementInterval = NORM_TICK_MIN;
// (Note this can limit our ramp up of data rate during slow start)
if (measurementInterval < NORM_TICK_MIN)
measurementInterval = NORM_TICK_MIN;
recv_accumulator += msgSize;
unsigned short oldRateQuantized = recv_rate_quantized;
if (interval > 0.0)
{
double currentRecvRate = ((double)(recv_accumulator)) / interval;
if ((interval >= measurementInterval) && (currentRecvRate < recv_rate))
{
// Make sure we've allowed sufficient time for a measurement at low rates
double minInterval = 4.0 * ((double)(MAX(nominal_packet_size, segment_size)))/ recv_rate;
if (measurementInterval < minInterval)
measurementInterval = minInterval;
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double minInterval = 4.0 * nominalSize / recv_rate;
if (measurementInterval < minInterval) measurementInterval = minInterval;
}
if (interval >= measurementInterval)
{
recv_rate = currentRecvRate; //((double)(recv_accumulator)) / interval;
recv_rate = currentRecvRate;
prev_update_time = currentTime;
recv_accumulator = msgSize;
recv_accumulator = 0;
}
else
else if (0.0 == recv_rate)
{
recv_accumulator += msgSize;
recv_rate = currentRecvRate;
}
recv_rate_quantized = NormQuantizeRate(recv_rate);
}
else if (0.0 == recv_rate)
{
// Approximate initial recv_rate when initial packets arrive in a burst
recv_rate = ((double)(recv_accumulator)) / NORM_TICK_MIN;
recv_rate_quantized = NormQuantizeRate(recv_rate);
}
nominal_packet_size += 0.05 * (((double)msgSize) - nominal_packet_size);
if (recv_rate_quantized != oldRateQuantized)
{
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double activityInterval =
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if ((activity_timer.GetInterval() > (NORM_ROBUST_FACTOR*grtt_estimate)) ||
(activityInterval > grtt_estimate))
{
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
if (activity_timer.IsActive()) activity_timer.Reschedule();
}
}
}
else
{
/*if (send_rate > 0.0)
recv_rate = send_rate;
else
recv_rate = ((double)msgSize) / grtt_estimate;*/
recv_rate = 0.0;
recv_rate = 0.0;
recv_rate_quantized = NormQuantizeRate(0.0);
prev_update_time = currentTime;
recv_accumulator = msgSize;
recv_accumulator = 0;
nominal_packet_size = msgSize;
}
nominal_packet_size += 0.05 * (((double)msgSize) - nominal_packet_size);
} // end NormServerNode::UpdateRecvRate()
void NormServerNode::Activate(bool isObjectMsg)
{
if (!activity_timer.IsActive())
{
activity_timer.SetInterval(grtt_estimate*NORM_ROBUST_FACTOR);
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double activityInterval =
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
session.ActivateTimer(activity_timer);
server_active = false;
session.Notify(NormController::REMOTE_SERVER_ACTIVE, this, NULL);
session.Notify(NormController::REMOTE_SENDER_ACTIVE, this, NULL);
}
else if (isObjectMsg)
{
@ -1757,7 +1830,7 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/)
DMSG(0, "NormServerNode::OnActivityTimeout() node>%lu server>%lu gone inactive?\n",
LocalNodeId(), GetId());
FreeBuffers();
session.Notify(NormController::REMOTE_SERVER_INACTIVE, this, NULL);
session.Notify(NormController::REMOTE_SENDER_INACTIVE, this, NULL);
}
else
{
@ -1788,18 +1861,12 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/)
}
else
{
RepairCheck(NormObject::THRU_OBJECT, // (TBD) thru object???
RepairCheck(NormObject::THRU_OBJECT,
max_pending_object, 0, 0);
//RepairCheck(NormObject::TO_BLOCK, // (TBD) thru object???
// max_pending_object, 0, 0);
}
}
// At low recv data rates, we adjust the activity timeout a little.
// to _help_ prevent premature timeout
double nominalInterval = nominal_packet_size / recv_rate;
if (grtt_estimate < nominalInterval)
activity_timer.SetInterval(NORM_ROBUST_FACTOR * nominalInterval);
}
server_active = false;
return true;
@ -1812,16 +1879,22 @@ bool NormServerNode::UpdateLossEstimate(const struct timeval& currentTime,
bool result = loss_estimator.Update(currentTime, seq, ecn);
if (result && slow_start)
{
// Calculate loss initialization based on current receive rate
// and rtt estimation
double lossInit = (recv_rate * rtt_estimate) /
(segment_size*sqrt(3.0/2.0));
(nominal_packet_size*sqrt(3.0/2.0));
lossInit = (lossInit*lossInit);
double currentInterval = (double)loss_estimator.LastLossInterval();
lossInit = 1.0 / MAX(lossInit, currentInterval);
loss_estimator.SetInitialLoss(lossInit);
// At this point, "LastLossInterval()" will always be non-zero
double altLoss = (double)loss_estimator.LastLossInterval();
double altInit = (altLoss > 0.0) ? (1.0 / altLoss) : 0.5;
if (altInit < lossInit)
loss_estimator.SetInitialLoss(altInit);
else
loss_estimator.SetInitialLoss(lossInit);
slow_start = false;
}
return result;
}
} // end NormServerNode::UpdateLossEstimate()
void NormServerNode::AttachCCFeedback(NormAckMsg& ack)
{
@ -1855,7 +1928,7 @@ void NormServerNode::AttachCCFeedback(NormAckMsg& ack)
// LocalNodeId(), 8.0e-03*NormUnquantizeRate(ext.GetCCRate()) ,
// rtt_estimate, ccLoss, nominal_packet_size, 8.0e-03*recv_rate, slow_start);
ext.SetCCSequence(cc_sequence);
} // end ormServerNode::AttachCCFeedback()
} // end NormServerNode::AttachCCFeedback()
bool NormServerNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
{
@ -1946,9 +2019,11 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
if (!is_clr && !is_plr && session.Address().IsMulticast())
{
// Install cc feedback holdoff
if (cc_timer.IsActive()) cc_timer.Deactivate();
cc_timer.SetInterval(grtt_estimate*backoff_factor);
session.ActivateTimer(cc_timer);
if (cc_timer.IsActive())
cc_timer.Reschedule();
else
session.ActivateTimer(cc_timer);
cc_timer.DecrementRepeatCount();
}
else if (cc_timer.IsActive())
@ -2533,7 +2608,7 @@ bool NormLossEstimator2::Update(const struct timeval& currentTime,
if (history[0] < 100000) history[0]++;
return newLossEvent;
} // end NormLossEstimator2::ProcessRecvPacket()
} // end NormLossEstimator2::Update()
double NormLossEstimator2::LossFraction()
{
@ -2612,7 +2687,7 @@ double NormLossEstimator2::TfrcLossFraction()
break;
}
}
double s0 = average / scaling;
double s0 = (average > 0.0) ? average / scaling : 0.0;
// Use max of old/new averages
return (1.0 / MAX(s0, s1));
} // end NormLossEstimator2::LossFraction()

6
common/normNode.h
View File

@ -238,6 +238,8 @@ class NormServerNode : public NormNode
bool UnicastNacks() {return unicast_nacks;}
void SetUnicastNacks(bool state) {unicast_nacks = state;}
double GetGrttEstimate() {return grtt_estimate;}
bool UpdateLossEstimate(const struct timeval& currentTime,
unsigned short theSequence,
bool ecnStatus = false);
@ -364,6 +366,9 @@ class NormServerNode : public NormNode
private:
static const double DEFAULT_NOMINAL_INTERVAL;
static const double ACTIVITY_INTERVAL_MIN;
bool PassiveRepairCheck(NormObjectId objectId,
NormBlockId blockId,
NormSegmentId segmentId);
@ -440,6 +445,7 @@ class NormServerNode : public NormNode
bool slow_start;
double send_rate; // sender advertised rate
double recv_rate; // measured recv rate
unsigned short recv_rate_quantized;
struct timeval prev_update_time; // for recv_rate measurement
unsigned long recv_accumulator; // for recv_rate measurement
double nominal_packet_size;

63
common/normObject.cpp
View File

@ -291,7 +291,8 @@ bool NormObject::HandleBlockRequest(NormBlockId nextId, NormBlockId lastId)
return increasedRepair;
} // end NormObject::HandleBlockRequest();
bool NormObject::TxReset(NormBlockId firstBlock)
bool NormObject::TxReset(NormBlockId firstBlock, bool requeue)
{
bool increasedRepair = false;
if (!pending_info && HaveInfo())
@ -319,6 +320,7 @@ bool NormObject::TxReset(NormBlockId firstBlock)
nparity,
session.ServerAutoParity(),
segment_size);
if (requeue) block->ClearFlag(NormBlock::IN_REPAIR); // since we're requeuing
}
}
return increasedRepair;
@ -539,7 +541,7 @@ bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd)
return true;
} // end NormObject::AppendRepairAdv()
// This is used by server for watermark check
// This is used by sender for watermark check
bool NormObject::FindRepairIndex(NormBlockId& blockId, NormSegmentId& segmentId) const
{
if (repair_info)
@ -1432,8 +1434,8 @@ NormBlock* NormObject::StealNewestBlock(bool excludeBlock, NormBlockId excludeId
} // end NormObject::StealNewestBlock()
// For silent client resource management, steals newer block resources when
// needing resources for ordinally older blocks.
// For silent client resource management, steals older block resources when
// needing resources for ordinally newer blocks.
NormBlock* NormObject::StealOldestBlock(bool excludeBlock, NormBlockId excludeId)
{
if (block_buffer.IsEmpty())
@ -1545,8 +1547,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
if (!block_buffer.Insert(block))
{
ASSERT(STREAM == type);
ASSERT(blockId > block_buffer.RangeLo());
//if (blockId > block_buffer.RangeLo())
if (blockId > block_buffer.RangeLo())
{
NormBlock* b = block_buffer.Find(block_buffer.RangeLo());
ASSERT(b);
@ -1555,13 +1556,26 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
bool success = block_buffer.Insert(block);
ASSERT(success);
}
/*else
else if (IsStream())
{
DMSG(0, "NormObject::NextServerMsg() node>%lu Warning! can't repair old block\n", LocalNodeId());
session.ServerPutFreeBlock(block);
pending_mask.Unset(blockId);
return false;
}*/
if (pending_mask.IsSet())
{
return NextServerMsg(msg);
}
else
{
static_cast<NormStreamObject*>(this)->StreamAdvance();
return false;
}
}
else
{
ASSERT(0);
return false;
}
}
}
NormSegmentId segmentId = 0;
@ -1587,17 +1601,9 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
}
data->SetPayloadLength(payloadLength);
/* stream payload flags have been deprecated in favor of "payload_msg_start" field
// Set FLAG_MSG_START as appropriated for NORM Protocol Version 1 compatibility
if (IsStream())
{
if (NormDataMsg::StreamPayloadFlagIsSet(buffer, NormDataMsg::FLAG_MSG_START))
data->SetFlag(NormObjectMsg::FLAG_MSG_START);
}
*/
// Perform incremental FEC encoding as needed
if ((block->ParityReadiness() == segmentId) && nparity)
// (TBD) && incrementalParity == true
// (TBD) && ((incrementalParity == true) || (auto_parity != 0))
{
// (TBD) for non-stream objects, catch alternate "last block/segment len"
// ZERO pad any "runt" segments before encoding
@ -1608,6 +1614,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
if (payloadLength < payloadMax)
memset(buffer+payloadLength, 0, payloadMax-payloadLength);
// (TBD) the encode routine could update the block's parity readiness
block->UpdateSegSizeMax(payloadLength);
session.ServerEncode(data->AccessPayload(), block->SegmentList(numData));
block->IncreaseParityReadiness();
}
@ -1622,7 +1629,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
ASSERT(block->ParityReady(numData));
char* segment = block->GetSegment(segmentId);
ASSERT(segment);
UINT16 payloadLength = segment_size;
UINT16 payloadLength = block->GetSegSizeMax(); // segment_size;
if (IsStream()) payloadLength += NormDataMsg::GetStreamPayloadHeaderLength();
data->SetPayload(segment, payloadLength);
}
@ -1698,6 +1705,7 @@ bool NormObject::CalculateBlockParity(NormBlock* block)
#endif // SIMULATE
if (payloadLength < payloadMax)
memset(buffer+payloadLength, 0, payloadMax-payloadLength+1);
block->UpdateSegSizeMax(payloadLength);
session.ServerEncode(buffer, block->SegmentList(numData));
}
else
@ -2397,6 +2405,7 @@ bool NormStreamObject::LockBlocks(NormBlockId firstId, NormBlockId lastId)
}
else
{
ASSERT(0);
return false;
}
nextId++;
@ -2631,7 +2640,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
{
DMSG(0, "NormStreamObject::WriteSegment() diff:%lu segmentOffset:%lu < read_offset:%lu \n",
diff, segmentOffset, read_offset);
ASSERT(0);
//ASSERT(0);
return false;
}
@ -2829,7 +2838,8 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool seekMsgStar
}
else
{
if (block_pool.GetCount() < block_pool_threshold)
if ((block_pool.GetCount() < block_pool_threshold) ||
(session.ReceiverIsLowDelay() && (read_index.block < max_pending_block)))
{
// Force read_index forward and try again.
if (++read_index.segment >= ndata)
@ -2863,7 +2873,8 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool seekMsgStar
}
else
{
if (block_pool.GetCount() < block_pool_threshold)
if ((block_pool.GetCount() < block_pool_threshold) ||
(session.ReceiverIsLowDelay() && (read_index.block < max_pending_block)))
{
// Force read_index forward and try again.
if (++read_index.segment >= ndata)
@ -3078,7 +3089,7 @@ void NormStreamObject::Terminate()
NormBlock* block = stream_buffer.Find(write_index.block);
if (!block)
{
if (!(block = block_pool.Get()))
if (NULL == (block = block_pool.Get()))
{
block = stream_buffer.Find(stream_buffer.RangeLo());
ASSERT(block);
@ -3281,11 +3292,11 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom)
msg_start = false;
}
UINT16 count = (UINT16)(len - nBytes);
UINT16 space = segment_size - index;
UINT32 count = len - nBytes;
UINT32 space = (UINT32)(segment_size - index);
count = MIN(count, space);
#ifdef SIMULATE
UINT16 simCount = index + NormDataMsg::GetStreamPayloadHeaderLength();
UINT32 simCount = index + NormDataMsg::GetStreamPayloadHeaderLength();
simCount = (simCount < SIM_PAYLOAD_MAX) ? (SIM_PAYLOAD_MAX - simCount) : 0;
simCount = MIN(count, simCount);
memcpy(segment+index+NormDataMsg::GetStreamPayloadHeaderLength(), buffer+nBytes, simCount);

5
common/normObject.h
View File

@ -167,7 +167,7 @@ class NormObject
NormBlock* ServerRecoverBlock(NormBlockId blockId);
bool CalculateBlockParity(NormBlock* block);
bool TxReset(NormBlockId firstBlock = NormBlockId(0));
bool TxReset(NormBlockId firstBlock = NormBlockId(0), bool requeue = false);
bool TxResetBlocks(NormBlockId nextId, NormBlockId lastId);
bool TxUpdateBlock(NormBlock* theBlock,
NormSegmentId firstSegmentId,
@ -452,6 +452,9 @@ class NormStreamObject : public NormObject
{block_pool_threshold = value;}
void ResetPostedTxQueueVacancy() {posted_tx_queue_vacancy = false;}
bool IsReadIndex(NormBlockId blockId, NormSegmentId segmentId) const
{return ((blockId == read_index.block) && (segmentId == read_index.segment));}
private:
void Terminate();

5
common/normSegment.cpp
View File

@ -124,7 +124,8 @@ bool NormBlock::Init(UINT16 totalSize)
size = totalSize;
erasure_count = 0;
parity_count = 0;
parity_offset = 0;;
parity_offset = 0;
seg_size_max = 0;
return true;
} // end NormBlock::Init()
@ -222,7 +223,7 @@ bool NormBlock::TxReset(UINT16 numData,
pending_mask.SetBits(0, numData+autoParity);
pending_mask.UnsetBits(numData+autoParity, numParity-autoParity);
parity_offset = autoParity; // reset parity since we're resending this one
parity_count = numParity; // no parity repair this repair cycle
parity_count = numParity; // no parity repair this repair cycle
SetFlag(IN_REPAIR);
if (!ParityReady(numData)) // (TBD) only when incrementalParity == true
{

7
common/normSegment.h
View File

@ -66,6 +66,7 @@ class NormBlock
void Destroy();
void SetFlag(NormBlock::Flag flag) {flags |= flag;}
void ClearFlag(NormBlock::Flag flag) {flags &= ~flag;}
bool InRepair() {return (0 != (flags & IN_REPAIR));}
bool ParityReady(UINT16 ndata) {return (erasure_count == ndata);}
UINT16 ParityReadiness() {return erasure_count;}
@ -109,6 +110,7 @@ class NormBlock
parity_count = 0;
parity_offset = autoParity;
flags = 0;
seg_size_max = 0;
}
void TxRecover(NormBlockId& blockId, UINT16 ndata, UINT16 nparity)
{
@ -119,11 +121,15 @@ class NormBlock
parity_count = nparity; // force recovered blocks to
parity_offset = nparity; // explicit repair mode ???
flags = IN_REPAIR;
seg_size_max = 0;
}
bool TxReset(UINT16 ndata, UINT16 nparity, UINT16 autoParity,
UINT16 segmentSize);
bool TxUpdate(NormSegmentId nextId, NormSegmentId lastId,
UINT16 ndata, UINT16 nparity, UINT16 erasureCount);
void UpdateSegSizeMax(UINT16 segSize)
{seg_size_max = (segSize > seg_size_max) ? segSize : seg_size_max;}
UINT16 GetSegSizeMax() {return seg_size_max;}
bool HandleSegmentRequest(NormSegmentId nextId, NormSegmentId lastId,
UINT16 ndata, UINT16 nparity,
@ -246,6 +252,7 @@ class NormBlock
UINT16 erasure_count;
UINT16 parity_count;
UINT16 parity_offset;
UINT16 seg_size_max;
ProtoBitmask pending_mask;
ProtoBitmask repair_mask;

107
common/normSession.cpp
View File

@ -35,8 +35,8 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
grtt_max(DEFAULT_GRTT_MAX),
grtt_decrease_delay_count(DEFAULT_GRTT_DECREASE_DELAY),
grtt_response(false), grtt_current_peak(0.0), grtt_age(0.0),
cc_enable(false), cc_sequence(0), cc_slow_start(true),
is_client(false), unicast_nacks(false), client_silent(false),
cc_enable(false), cc_sequence(0), cc_slow_start(true), cc_active(false),
is_client(false), unicast_nacks(false), client_silent(false), receiver_low_delay(false),
default_repair_boundary(NormServerNode::BLOCK_BOUNDARY),
default_nacking_mode(NormObject::NACK_NORMAL),
trace(false), tx_loss_rate(0.0), rx_loss_rate(0.0),
@ -219,7 +219,7 @@ void NormSession::Close()
if (address.IsMulticast())
{
const char* interfaceName = ('\0' != interface_name[0]) ?
interface_name : NULL;
interface_name : NULL;
rx_socket.LeaveGroup(address, interfaceName);
}
rx_socket.Close();
@ -295,12 +295,14 @@ void NormSession::SetTxRateInternal(double txRate)
grtt_advertised = NormUnquantizeRtt(grtt_quantized);
}
if (grttQuantizedOld != grtt_quantized)
{
DMSG(4, "NormSession::SetTxRateInternal() node>%lu %s to new grtt to: %lf sec\n",
LocalNodeId(),
(grttQuantizedOld < grtt_quantized) ? "increased" : "decreased",
grtt_advertised);
Notify(NormController::GRTT_UPDATED, (NormServerNode*)NULL, (NormObject*)NULL);
}
}
} // end NormSession::SetTxRateInternal()
@ -434,9 +436,12 @@ bool NormSession::StartServer(UINT16 instanceId,
}
if ((tx_rate_max >= 0.0) && (tx_rate > tx_rate_max))
txRate = tx_rate_max;
tx_rate = txRate; // keep grtt at initial
//SetTxRateInternal(txRate); // adjusts grtt_advertised as needed
//tx_rate = txRate; // keep grtt at initial
SetTxRateInternal(txRate); // adjusts grtt_advertised as needed
}
cc_slow_start = true;
cc_active = false;
grtt_age = 0.0;
probe_pending = false;
probe_data_check = false;
@ -676,6 +681,8 @@ void NormSession::Serve()
posted_tx_queue_empty = true; // repair-delayed stream advance
else
tx_pending_mask.Unset(obj->GetId());
// (TBD) post TX_OBJECT_SENT notification
}
else if (obj->IsStream())
{
@ -1161,6 +1168,7 @@ NormStreamObject* NormSession::QueueTxStream(UINT32 bufferSize,
}
} // end NormSession::QueueTxStream()
#ifdef SIMULATE
NormSimObject* NormSession::QueueTxSim(unsigned long objectSize)
{
@ -1241,6 +1249,36 @@ bool NormSession::QueueTxObject(NormObject* obj)
return true;
} // end NormSession::QueueTxObject()
bool NormSession::RequeueTxObject(NormObject* obj)
{
ASSERT(obj);
if (obj->IsStream())
{
// (TBD) allow buffered stream to be reset?
DMSG(0, "NormSession::RequeueTxObject() error: can't requeue NORM_OBJECT_STREAM\n");
return false;
}
NormObjectId objectId = obj->GetId();
if (tx_table.Find(objectId) == obj)
{
if (tx_pending_mask.Set(objectId))
{
obj->TxReset(0, true);
return true;
}
else
{
DMSG(0, "NormSession::RequeueTxObject() error: couldn't set object as pending\n");
return false;
}
}
else
{
DMSG(0, "NormSession::RequeueTxObject() error: couldn't find object\n");
return false;
}
} // end NormSession::RequeueTxObject()
void NormSession::DeleteTxObject(NormObject* obj)
{
ASSERT(obj);
@ -1644,7 +1682,7 @@ void NormSession::ClientHandleObjectMessage(const struct timeval& currentTime,
{
if ((theServer = new NormServerNode(*this, msg.GetSourceId())))
{
Notify(NormController::REMOTE_SERVER_NEW, theServer, NULL);
Notify(NormController::REMOTE_SENDER_NEW, theServer, NULL);
if (theServer->Open(msg.GetInstanceId()))
{
server_tree.AttachNode(theServer);
@ -1667,11 +1705,12 @@ void NormSession::ClientHandleObjectMessage(const struct timeval& currentTime,
}
}
theServer->Activate(true);
theServer->UpdateLossEstimate(currentTime, msg.GetSequence());
theServer->SetAddress(msg.GetSource());
theServer->UpdateRecvRate(currentTime, msg.GetLength());
theServer->UpdateLossEstimate(currentTime, msg.GetSequence());
theServer->IncrementRecvTotal(msg.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG
theServer->HandleObjectMessage(msg);
theServer->UpdateRecvRate(currentTime, msg.GetLength());
} // end NormSession::ClientHandleObjectMessage()
void NormSession::ClientHandleCommand(const struct timeval& currentTime,
@ -1701,7 +1740,7 @@ void NormSession::ClientHandleCommand(const struct timeval& currentTime,
// LocalNodeId());
if ((theServer = new NormServerNode(*this, cmd.GetSourceId())))
{
Notify(NormController::REMOTE_SERVER_NEW, theServer, NULL);
Notify(NormController::REMOTE_SENDER_NEW, theServer, NULL);
if (theServer->Open(cmd.GetInstanceId()))
{
server_tree.AttachNode(theServer);
@ -1724,11 +1763,11 @@ void NormSession::ClientHandleCommand(const struct timeval& currentTime,
}
}
theServer->Activate(false);
theServer->UpdateLossEstimate(currentTime, cmd.GetSequence());
theServer->SetAddress(cmd.GetSource());
theServer->IncrementRecvTotal(cmd.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG
theServer->HandleCommand(currentTime, cmd);
theServer->UpdateRecvRate(currentTime, cmd.GetLength());
theServer->UpdateLossEstimate(currentTime, cmd.GetSequence());
theServer->IncrementRecvTotal(cmd.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG
theServer->HandleCommand(currentTime, cmd);
} // end NormSession::ClientHandleCommand()
@ -1737,26 +1776,26 @@ double NormSession::CalculateRtt(const struct timeval& currentTime,
{
if (grttResponse.tv_sec || grttResponse.tv_usec)
{
double clientRtt;
// Calculate rtt estimate for this client and process the response
double rcvrRtt;
// Calculate rtt estimate for this receiver and process the response
if (currentTime.tv_usec < grttResponse.tv_usec)
{
clientRtt =
rcvrRtt =
(double)(currentTime.tv_sec - grttResponse.tv_sec - 1);
clientRtt +=
rcvrRtt +=
((double)(1000000 - (grttResponse.tv_usec - currentTime.tv_usec))) / 1.0e06;
}
else
{
clientRtt =
rcvrRtt =
(double)(currentTime.tv_sec - grttResponse.tv_sec);
clientRtt +=
rcvrRtt +=
((double)(currentTime.tv_usec - grttResponse.tv_usec)) / 1.0e06;
}
// Lower limit on RTT (because of coarse timer resolution on some systems,
// this can sometimes actually end up a negative value!)
// (TBD) this should be system clock granularity?
return (clientRtt < 1.0e-06) ? 1.0e-06 : clientRtt;
return (rcvrRtt < 1.0e-06) ? 1.0e-06 : rcvrRtt;
}
else
{
@ -2390,7 +2429,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
else
{
// Entire block already tx pending, don't recover
DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu "
DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu "
"recvd SEGMENT repair request for pending block.\n");
break;
}
@ -2812,8 +2851,8 @@ bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/)
LocalNodeId(), (UINT16)objectId);
if (obj->ActivateRepairs())
{
DMSG(6, "NormSession::OnRepairTimeout() node>%lu activated obj>%hu repairs fd=%d...\n",
LocalNodeId(), (UINT16)objectId, static_cast<NormFileObject*>(obj)->file.fd);
DMSG(6, "NormSession::OnRepairTimeout() node>%lu activated obj>%hu repairs ...\n",
LocalNodeId(), (UINT16)objectId);
if (!tx_pending_mask.Set(objectId))
DMSG(0, "NormSession::OnRepairTimeout() rx_pending_mask.Set(%hu) error (2)\n",
(UINT16)objectId);
@ -3359,6 +3398,11 @@ void NormSession::AdjustRate(bool onResponse)
double txRate = tx_rate;
if (onResponse)
{
if (!cc_active)
{
cc_active = true;
Notify(NormController::CC_ACTIVE, NULL, NULL);
}
if (data_active) // adjust only if actively transmitting
{
// Adjust rate based on CLR feedback and
@ -3446,7 +3490,19 @@ void NormSession::AdjustRate(bool onResponse)
if (minRate > ((double)(segment_size)))
minRate = (double)(segment_size);
}
if (txRate < minRate) txRate = minRate;
if (txRate <= minRate)
{
txRate = minRate;
if ((NULL == clr) || (!clr->IsActive()))
{
// Post notification that no cc feedback is being received
if (cc_active)
{
cc_active = false;
Notify(NormController::CC_INACTIVE, NULL, NULL);
}
}
}
if ((tx_rate_max >= 0.0) && (txRate > tx_rate_max))
txRate = tx_rate_max;
if (txRate != tx_rate) SetTxRateInternal(txRate);
@ -3569,7 +3625,8 @@ NormSession* NormSessionMgr::NewSession(const char* sessionAddress,
GetErrorString());
return ((NormSession*)NULL);
}
theSession->SetAddress(theAddress);
theSession->SetAddress(theAddress);
theSession->SetTxPort(sessionPort); /* JPH 4/24/06 */
// Add new session to our session list
theSession->next = top_session;
top_session = theSession;

40
common/normSession.h
View File

@ -21,22 +21,25 @@ class NormController
TX_WATERMARK_COMPLETED,
TX_OBJECT_SENT,
TX_OBJECT_PURGED,
LOCAL_SERVER_CLOSED,
REMOTE_SERVER_NEW,
REMOTE_SERVER_ACTIVE,
REMOTE_SERVER_INACTIVE,
REMOTE_SERVER_PURGED,
LOCAL_SENDER_CLOSED,
REMOTE_SENDER_NEW,
REMOTE_SENDER_ACTIVE,
REMOTE_SENDER_INACTIVE,
REMOTE_SENDER_PURGED,
RX_OBJECT_NEW,
RX_OBJECT_INFO,
RX_OBJECT_UPDATED,
RX_OBJECT_COMPLETED,
RX_OBJECT_ABORTED
RX_OBJECT_ABORTED,
GRTT_UPDATED,
CC_ACTIVE, // posted when cc feedback is detected
CC_INACTIVE // posted when no cc feedback and min rate reached
};
virtual void Notify(NormController::Event event,
class NormSessionMgr* sessionMgr,
class NormSession* session,
class NormServerNode* server,
class NormServerNode* sender,
class NormObject* object) = 0;
}; // end class NormController
@ -59,11 +62,11 @@ class NormSessionMgr
void Notify(NormController::Event event,
class NormSession* session,
class NormServerNode* server,
class NormServerNode* sender,
class NormObject* object)
{
if (controller)
controller->Notify(event, this, session, server, object);
controller->Notify(event, this, session, sender, object);
}
void ActivateTimer(ProtoTimer& timer) {timer_mgr.ActivateTimer(timer);}
@ -220,6 +223,9 @@ class NormSession
UINT32 dataLen,
const char* infoPtr = NULL,
UINT16 infoLen = 0);
bool RequeueTxObject(NormObject* obj);
void DeleteTxObject(NormObject* obj);
// postive ack mgmnt
@ -307,9 +313,17 @@ class NormSession
{return remote_server_buffer_size;}
void ClientSetUnicastNacks(bool state) {unicast_nacks = state;}
bool ClientGetUnicastNacks() const {return unicast_nacks;}
void ClientSetSilent(bool state) {client_silent = state;}
void ClientSetSilent(bool state)
{
client_silent = state;
receiver_low_delay = client_silent ? receiver_low_delay : false;
}
bool ClientIsSilent() const {return client_silent;}
void ReceiverSetLowDelay(bool state)
{receiver_low_delay = client_silent ? state : false;}
bool ReceiverIsLowDelay() {return receiver_low_delay;}
NormObject::NackingMode ClientGetDefaultNackingMode() const
{return default_nacking_mode;}
void ClientSetDefaultNackingMode(NormObject::NackingMode nackingMode)
@ -473,21 +487,23 @@ class NormSession
double gsize_advertised;
UINT8 gsize_quantized;
// Server congestion control parameters
// Sender congestion control parameters
bool cc_enable;
UINT8 cc_sequence;
NormNodeList cc_node_list;
bool cc_slow_start;
bool cc_active;
unsigned long sent_accumulator; // for sentRate measurement
double nominal_packet_size;
bool data_active; // true when actively sending data
// Client parameters
// Receiver parameters
bool is_client;
NormNodeTree server_tree;
unsigned long remote_server_buffer_size;
bool unicast_nacks;
bool client_silent;
bool receiver_low_delay;
NormServerNode::RepairBoundary default_repair_boundary;
NormObject::NackingMode default_nacking_mode;

25
common/normSimAgent.cpp
View File

@ -48,7 +48,7 @@ const char* const NormSimAgent::cmd_list[] =
"-trace", // message tracing
"+txloss", // tx packet loss percent
"+rxloss", // rx packet loss percent
"+address", // session dest addres
"+address", // session dest address
"+ttl", // multicast ttl
"+rate", // tx rate
"+cc", // congestion control on/off
@ -67,12 +67,12 @@ const char* const NormSimAgent::cmd_list[] =
"+grtt", // grtt estimate
"+txbuffer", // tx buffer size (bytes)
"+rxbuffer", // rx buffer size (bytes)
"+start", // open session and beging activity
"+start", // open session and begin activity
"-stop", // cease activity and close session
"+sendFile", // queue a "sim" file of <size> bytes for transmission
"+sendRandomFile", // queue random-size file size range <sizeMin>:<sizeMax>
"+sendStream", // send a simulated NORM stream
"+openStream", // open a stream object for messaging
"+openStream", // open a stream object for messaging (sending)
"+push", // "on" means real-time push stream advancement (non-blocking)
"+flush", // stream flush mode
"-doFlush", // invoke flushing of stream
@ -691,7 +691,7 @@ void NormSimAgent::Notify(NormController::Event event,
}
break;
case NormObject::DATA:
DMSG(0, "NormSimAgent::Notify() FILE/DATA objects not _yet_ supported...\n");
DMSG(0, "NormSimAgent::Notify() DATA objects not _yet_ supported...\n");
break;
default:
DMSG(0, "NormSimAgent::Notify() INVALID object type!\n");
@ -733,8 +733,8 @@ void NormSimAgent::Notify(NormController::Event event,
unsigned int want = 2 - mgen_bytes;
unsigned int got = want;
bool findMsgSync = msg_sync ? false : true;
if (((NormStreamObject*)object)->Read(mgen_buffer + mgen_bytes,
&got, findMsgSync))
if ((static_cast<NormStreamObject*>(object))->Read(mgen_buffer + mgen_bytes,
&got, findMsgSync))
{
mgen_bytes += got;
msg_sync = true;
@ -761,8 +761,8 @@ void NormSimAgent::Notify(NormController::Event event,
{
unsigned int want = MIN(mgen_pending_bytes, 62);
unsigned int got = want;
if (((NormStreamObject*)object)->Read(mgen_buffer+mgen_bytes,
&got))
if ((static_cast<NormStreamObject*>(object))->Read(mgen_buffer+mgen_bytes,
&got))
{
mgen_pending_bytes -= got;
mgen_bytes += got;
@ -781,7 +781,7 @@ void NormSimAgent::Notify(NormController::Event event,
char buffer[256];
unsigned int want = MIN(256, mgen_pending_bytes);
unsigned int got = want;
if (((NormStreamObject*)object)->Read(buffer, &got))
if ((static_cast<NormStreamObject*>(object))->Read(buffer, &got))
{
mgen_pending_bytes -= got;
mgen_bytes += got;
@ -800,7 +800,12 @@ void NormSimAgent::Notify(NormController::Event event,
ProtoAddress srcAddr;
srcAddr.ResolveFromString(server->GetAddress().GetHostString());
srcAddr.SetPort(server->GetAddress().GetPort());
#ifdef OPNET // JPH 4/11/06 Use packet stream to send msg to mgen rather than direct call on mgen method.
HandleMgenMessage(mgen_buffer, mgen_bytes, srcAddr);
#else
mgen->HandleMgenMessage(mgen_buffer, mgen_bytes, srcAddr);
#endif OPNET
mgen_bytes = 0;
}
} // end if (mgen_pending_bytes)
@ -813,7 +818,7 @@ void NormSimAgent::Notify(NormController::Event event,
unsigned int got = want;
while (1)
{
if (((NormStreamObject*)object)->Read(buffer, &got))
if ((static_cast<NormStreamObject*>(object))->Read(buffer, &got))
{
// Break when data is no longer available
if (got != want) break;

10
common/normSimAgent.h
View File

@ -15,8 +15,8 @@ class NormSimAgent : public NormController
bool ProcessCommand(const char* cmd, const char* val);
// Note: don't allow client _and_ server operation at same time
bool StartServer();
bool StartClient();
bool StartServer(); // start sender
bool StartClient(); // start receiver
bool IsActive() {return (NULL != session);}
void Stop();
@ -29,6 +29,10 @@ class NormSimAgent : public NormController
enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG};
CmdType CommandType(const char* cmd);
virtual unsigned long GetAgentId() = 0;
/* JPH 4/11/06 Use packet stream instead of direct call on mgen process */
void HandleMgenMessage(char* buffer,
unsigned int len,
const ProtoAddress& srcAddr);
private:
void OnInputReady();
@ -83,7 +87,7 @@ class NormSimAgent : public NormController
char* tx_msg_buffer;
unsigned int tx_msg_len;
unsigned int tx_msg_index;
Mgen* mgen;
Mgen* mgen; // mgen receiver
char mgen_buffer[64];
bool msg_sync;
unsigned int mgen_bytes;

102
common/normTest.cpp
View File

@ -61,8 +61,8 @@ int main(int argc, char* argv[])
// Create a NORM session instance
NormSessionHandle session = NormCreateSession(instance,
"224.1.2.3",
6003,
"224.1.1.1",
6001,
localId);
//NormSetTOS(session, 0x20);
@ -72,7 +72,7 @@ int main(int argc, char* argv[])
// Uncomment to turn on debug NORM message tracing
//NormSetMessageTrace(session, true);
// Uncomment to turn on some random packet loss
NormSetTxLoss(session, 25.0); // 10% packet loss
//NormSetTxLoss(session, 10.0); // 10% packet loss
//NormSetRxLoss(session, 20.0);
struct timeval currentTime;
ProtoSystemTime(currentTime);
@ -82,12 +82,12 @@ int main(int argc, char* argv[])
NormSetGrttEstimate(session, 0.001); // 1 msec initial grtt
NormSetTransmitRate(session, 256.0e+04); // in bits/second
NormSetTransmitRate(session, 1.0e+06); // in bits/second
// Uncomment to enable TCP-friendly congestion control (overrides NormSetTransmitRate())
//NormSetCongestionControl(session, true);
NormSetCongestionControl(session, true);
NormSetTransmitRateBounds(session, 1000.0, -1.0);
//NormSetTransmitRateBounds(session, 1000.0, -1.0);
NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK);
@ -96,21 +96,20 @@ int main(int argc, char* argv[])
// is _not_ recommended when unicast feedback may be
// possible!)
//NormSetTxPort(session, 6001);
// Uncomment to allow reuse of rx port
// (This allows multiple "normTest" instances on the same machine
// for the same NormSession - note those instances must use
// unique local NormNodeIds (see NormCreateSession() above).
NormSetRxPortReuse(session, true);
//NormSetRxPortReuse(session, true);
// Uncomment to receive your own traffic
NormSetLoopback(session, true);
//NormSetLoopback(session, true);
//NormSetSilentReceiver(session, true);
// Uncomment this line to participate as a receiver
NormStartReceiver(session, 1024*1024);
//NormStartReceiver(session, 1024*1024);
// Uncomment to set large rx socket buffer size
// (might be needed for high rate sessions)
@ -135,8 +134,10 @@ int main(int argc, char* argv[])
//const char* filePath = "/home/adamson/pkgs/rh73.tgz";
const char* fileName = "file1.jpg";
const char* fileName2 = "file2.jpg";
// Uncomment this line to send a stream instead of the file
stream = NormStreamOpen(session, 2*1024*1024);
stream = NormStreamOpen(session, 4*1024*1024);
// NORM_FLUSH_PASSIVE automatically flushes full writes to
// the stream.
@ -148,7 +149,7 @@ int main(int argc, char* argv[])
int txLen = 0;
int rxIndex = 0;
#define STREAM_MSG_PREFIX_SIZE 1024 // we pad with 'a' character value (the msgPrefix is a prefix)
#define STREAM_MSG_PREFIX_SIZE 1000 // we pad with 'a' character value (the msgPrefix is a prefix)
char msgPrefix[STREAM_MSG_PREFIX_SIZE];
memset(msgPrefix, 'a', STREAM_MSG_PREFIX_SIZE);
bool msgSync = false;
@ -156,7 +157,11 @@ int main(int argc, char* argv[])
int msgCount = 0;
int recvCount = -1; // used to monitor reliable stream reception
int sendCount = 0;
int sendMax = 300;//-1; // -1 means unlimited
int sendMax = -1;//300;//-1; // -1 means unlimited
int fileMax = 1;
NormObjectHandle txFile = NORM_OBJECT_INVALID;
NormEvent theEvent;
while (NormGetNextEvent(instance, &theEvent))
@ -165,8 +170,6 @@ int main(int argc, char* argv[])
{
case NORM_TX_QUEUE_VACANCY:
case NORM_TX_QUEUE_EMPTY:
fprintf(stderr, "stopping ....\n");
NormStopInstance(instance);
/*if (NORM_TX_QUEUE_VACANCY == theEvent.type)
TRACE("NORM_TX_QUEUE_VACANCY ...\n");
else
@ -216,38 +219,56 @@ int main(int argc, char* argv[])
}
else if (NORM_OBJECT_INVALID == stream) // we weren't sending a stream that finished
{
const char* namePtr = (0 == (sendCount & 0x01)) ? fileName : fileName2;
NormObjectHandle txFile =
NormFileEnqueue(session,
filePath,
namePtr,
strlen(fileName));
// Repeatedly queue our file for sending
if (NORM_OBJECT_INVALID == txFile)
if ((fileMax < 0) || (sendCount < fileMax))
{
DMSG(0, "normTest: error queuing file: %s\n", filePath);
break;
}
else
{
TRACE("QUEUED FILE ...\n");
sendCount++;
if (sendCount < 2) NormSetWatermark(session, txFile);
const char* namePtr = (0 == (sendCount & 0x01)) ? fileName : fileName2;
txFile =
NormFileEnqueue(session,
filePath,
namePtr,
strlen(fileName));
// Repeatedly queue our file for sending
if (NORM_OBJECT_INVALID == txFile)
{
DMSG(0, "normTest: error queuing file: %s\n", filePath);
break;
}
else
{
TRACE("QUEUED FILE ...\n");
sendCount++;
if (sendCount < 2) NormSetWatermark(session, txFile);
}
}
}
break;
case NORM_TX_OBJECT_PURGED:
DMSG(2, "normTest: NORM_TX_OBJECT_PURGED event ...\n");
break;
case NORM_TX_FLUSH_COMPLETED:
DMSG(2, "normTest: NORM_TX_FLUSH_COMPLETED event ...\n");
// This line of code "requeues" the last file sent
// (Good for use with silent receivers ... can repeat transmit object this way)
NormRequeueObject(session, txFile);
break;
case NORM_TX_WATERMARK_COMPLETED:
DMSG(2, "normTest: NORM_TX_WATERMARK_COMPLETED event ...\n");
break;
case NORM_TX_OBJECT_SENT:
DMSG(2, "normTest: NORM_TX_WATERMARK_COMPLETED event ...\n");
break;
case NORM_TX_OBJECT_PURGED:
DMSG(2, "normTest: NORM_TX_OBJECT_PURGED event ...\n");
break;
case NORM_CC_ACTIVE:
DMSG(2, "normTest: NORM_CC_ACTIVE event ...\n");
break;
case NORM_CC_INACTIVE:
DMSG(2, "normTest: NORM_CC_INACTIVE event ...\n");
break;
case NORM_RX_OBJECT_NEW:
DMSG(3, "normTest: NORM_RX_OBJECT_NEW event ...\n");
@ -392,24 +413,27 @@ int main(int argc, char* argv[])
case NORM_RX_OBJECT_ABORTED:
TRACE("normTest: NORM_RX_OBJECT_ABORTED event ...\n");
break;
case NORM_GRTT_UPDATED:
//TRACE("normTest: NORM_GRTT_UPDATED event ...\n");
break;
default:
TRACE("Got event type: %d\n", theEvent.type);
} // end switch(theEvent.type)
// Uncomment to exit program after sending "sendMax" messages or files
//if ((sendMax > 0) && (sendCount >= sendMax)) break;
if ((sendMax > 0) && (sendCount >= sendMax)) break;
} // end while (NormGetNextEvent())
fprintf(stderr, "normTest shutting down in 30 sec ...\n");
#ifdef WIN32
Sleep(30000);
#else
sleep(30); // allows time for cleanup if we're sending to someone else
#endif // if/else WIN32/UNIX
NormStreamClose(stream);
//NormStreamClose(stream); // stream is already closed in loop above
NormStopReceiver(session);
NormStopSender(session);
NormDestroySession(session);

2
common/normVersion.h
View File

@ -32,6 +32,6 @@
#ifndef _NORM_VERSION
#define _NORM_VERSION
#define VERSION "1.3b7"
#define VERSION "1.3b8"
#endif // _NORM_VERSION

46
opnet/OpnetNormProcess.h Normal file
View File

@ -0,0 +1,46 @@
#ifndef _OPNET_NORM_PROCESS
#define _OPNET_NORM_PROCESS
// JPH NORM 11/21/2005
#include "opnet.h"
#include "opnetProtoSimProcess.h"
#include "normSimAgent.h" // this includes protokit.h
class OpnetNormProcess : public OpnetProtoSimProcess, public NormSimAgent, public MgenSink
{
public:
OpnetNormProcess();
~OpnetNormProcess();
// OpnetProtoSimProcess's base class overrides
bool OnStartup(int argc, const char*const* argv);
bool ProcessCommands(int argc, const char*const* argv);
void OnShutdown();
void ReceivePacketMonitor(Ici* ici, Packet* pkt);
void TransmitPacketMonitor(Ici* ici, Packet* pkt);
// JPH 5/25/06 - backoff from mgen 42b8 to 42b6
// bool SendMgenMessage(MgenMsg& theMsg,
// bool checksum_enable,
// char* txBuffer);
bool SendMgenMessage(const char* txBuffer,
unsigned int len,
const ProtoAddress& dstAddr);
// NormSimAgent override
unsigned long GetAgentId() {return (unsigned long)addr();}
private:
IpT_Address addr();
}; // end class OpnetNormProcess
#endif // _OPNET_NORM_PROCESS

55
opnet/mgen-ns2_2node_norm-ip_routes_olsr.gdf Normal file
View File

@ -0,0 +1,55 @@
# This file contains the IP routing tables built by various routing protocols in
# the network model.
# It is intended to be used as a mechanism whereby dynamic routing information
# is recorded and reused to avoid running IP routing protocols in multiple
# simulations of the same network. This feature should be used only when router
# connectivities in the network (and hence their routing tables) do not change
# during the course of the simulation.
# This file is produced each time a simulation is run with the simulation
# attribute "Routing Table Export/Import" set to "Export". Multiple runs for
# the same network model will successively overwrite this file.
# The contents of this file can be imported into the network model by running a
# simulation with the simulation attribute "Routing Table Export/Import" to
# "Import".
# In order for this to work correctly, there should be no change in the network
# whose simulation produced this file, and the network that is importing it
# contents. In other words, the "Export" scenario and the "Import" scenario
# should be the same.
# Warning: Modification of this file by the user can lead to unexpected simulation
# results.
START_ROUTING_TABLE at 60.000000
#Module Object ID,Table Size,Number of Interfaces,Is static
508,5,0,0
#Module Hierarchical Name:
Campus Network.node_(0).olsr_protolib
#Interface Information: Interface,IP Address,Mask
0,-1,192.0.0.100,255.255.255.0
#OLSR Routing Table Contents:
#-----------------------------------------------------------------
# Dest Network,Dest Net Mask, Metric, Next Hop Addr,
#-----------------------------------------------------------------
192.0.0.101,255.255.255.255,1,192.0.0.101
END_ROUTING_TABLE
START_ROUTING_TABLE at 60.000000
#Module Object ID,Table Size,Number of Interfaces,Is static
1044,6,0,0
#Module Hierarchical Name:
Campus Network.node_(1).olsr_protolib
#Interface Information: Interface,IP Address,Mask
0,-1,192.0.0.101,255.255.255.0
#OLSR Routing Table Contents:
#-----------------------------------------------------------------
# Dest Network,Dest Net Mask, Metric, Next Hop Addr,
#-----------------------------------------------------------------
192.0.0.100,255.255.255.255,1,192.0.0.100
END_ROUTING_TABLE

1540
opnet/norm_protolib.pr.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

BIN
opnet/norm_protolib.pr.m Normal file
View File

Binary file not shown.

BIN
win32/Norm.opt
View File

Binary file not shown.

BIN
win32/Norm.suo
View File

Binary file not shown.

3
win32/NormDll.vcproj
View File

@ -24,7 +24,7 @@
MinimalRebuild="TRUE"
BasicRuntimeChecks="3"
RuntimeLibrary="3"
UsePrecompiledHeader="3"
UsePrecompiledHeader="2"
WarningLevel="3"
Detect64BitPortabilityProblems="TRUE"
DebugInformationFormat="4"/>
@ -32,6 +32,7 @@
Name="VCCustomBuildTool"/>
<Tool
Name="VCLinkerTool"
AdditionalDependencies="ws2_32.lib iphlpapi.lib"
OutputFile="$(OutDir)/Norm.dll"
LinkIncremental="2"
GenerateDebugInformation="TRUE"

8
win32/NormLib.vcproj
View File

@ -3,7 +3,6 @@
ProjectType="Visual C++"
Version="7.10"
Name="NormLib"
ProjectGUID="{CEC5446E-BC6D-40B2-B496-78A211A9C1E6}"
SccProjectName=""
SccLocalPath="">
<Platforms>
@ -29,7 +28,7 @@
UsePrecompiledHeader="2"
PrecompiledHeaderFile=".\Debug/NormLib.pch"
AssemblerListingLocation=".\Debug/"
ObjectFile="Norm.lib"
ObjectFile=".\Debug/"
ProgramDataBaseFileName=".\Debug/"
WarningLevel="3"
SuppressStartupBanner="TRUE"
@ -38,7 +37,7 @@
Name="VCCustomBuildTool"/>
<Tool
Name="VCLibrarianTool"
OutputFile="$(OutDir)/NormLib.lib"
OutputFile=".\Debug\NormLib.lib"
SuppressStartupBanner="TRUE"/>
<Tool
Name="VCMIDLTool"/>
@ -81,6 +80,7 @@
UsePrecompiledHeader="2"
PrecompiledHeaderFile=".\Release/NormLib.pch"
AssemblerListingLocation=".\Release/"
ObjectFile=".\Release/"
ProgramDataBaseFileName=".\Release/"
WarningLevel="3"
SuppressStartupBanner="TRUE"
@ -89,7 +89,7 @@
Name="VCCustomBuildTool"/>
<Tool
Name="VCLibrarianTool"
OutputFile="$(OutDir)/NormLib.lib"
OutputFile=".\Release\NormLib.lib"
SuppressStartupBanner="TRUE"/>
<Tool
Name="VCMIDLTool"/>

4
win32/normTest.vcproj
View File

@ -24,7 +24,7 @@
Name="VCCLCompilerTool"
Optimization="0"
AdditionalIncludeDirectories="..\common,..\protolib\common,..\protolib\win32"
PreprocessorDefinitions="NORM_USE_DLL, PROTO_DEBUG, WIN32"
PreprocessorDefinitions="NORM_USE_DLL, PROTO_DEBUG, HAVE_ASSERT, WIN32"
BasicRuntimeChecks="3"
RuntimeLibrary="3"
BufferSecurityCheck="TRUE"
@ -41,7 +41,7 @@
Name="VCCustomBuildTool"/>
<Tool
Name="VCLinkerTool"
AdditionalDependencies="iphlpapi.lib ws2_32.lib odbc32.lib odbccp32.lib Norm.dll"
AdditionalDependencies="iphlpapi.lib ws2_32.lib odbc32.lib odbccp32.lib"
OutputFile=".\Debug/normTest.exe"
LinkIncremental="0"
SuppressStartupBanner="TRUE"

2
win32/win32PostProcess.cpp
View File

@ -276,7 +276,7 @@ bool Win32PostProcessor::ProcessFile(const char *path)
}
else
{
post_processor_handle = exeInfo.hProcess;
post_processor_handle = exeInfo.hProcess;
return true;
}
} // end Win32PostProcessor::ProcessFile()