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

update to latest

pull/2/head
Jeff Weston 2019-09-11 12:29:48 -04:00
parent 7dc0cea1b0
commit 120989e3d6
39 changed files with 1704 additions and 299 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
README.md
View File

@ -38,7 +38,7 @@ purposes, the authors would prefer that the code not be
re-distributed so that users will obtain the latest (and most
debugged/improved) release from the official NORM web site:
<http://cs.itd.nrl.navy.mil/work/norm>
<https://www.nrl.navy.mil/itd/ncs/products/norm>
SOURCE CODE
@ -102,14 +102,12 @@ NormUserGuide.pdf - Guide to "norm" demo app usage
VERSION.TXT - NORM version history notes
README.TXT - this file
README.md - this file
NOTES:
======
The NORM code depends upon the current "Protolib" release. The NORM
source code tarballs contain an appropriate release of "Protolib" as
part of the NORM source tree. If the NORM code is checked out from
our CVS server, it is necessary to also check out "protolib" separately
and provide paths to it for NORM to build.
The NORM code depends upon the current "Protolib" release:
https://www.nrl.navy.mil/itd/ncs/products/protolib
https://github.com/USNavalResearchLaboratory/protolib

5
VERSION.TXT
View File

@ -1,9 +1,10 @@
NORM Version History
Version 1.5.8
=============
- Corrections to eliminate compiler warnings and suport kfreebsd builds
(Thanks to Luca Buccassi)
- Modified waf build to build pkg-config file.cThanks to Luca Boccassi
for this and some other patches for Gnu/Hurd, etc.)
Version 1.5.7
=============

20
android/README.TXT Normal file
View File

@ -0,0 +1,20 @@
This directoy contains Android Studio project files for building the NORM
library for Android.
TO BUILD:
./gradlew build
The lib/build/ directory will contain the binary libraries built.
The "./gradlew clean" command can be used to delete the files built.
This directory may also be opened as an Android Studio project and built from
the IDE GUI.
This will evolve as we update NORM and associated projects to the newer Android
Studio tool chain.

16
android/build.gradle Normal file
View File

@ -0,0 +1,16 @@
buildscript {
repositories {
google()
jcenter()
}
dependencies {
classpath 'com.android.tools.build:gradle:3.1.1'
}
}
allprojects {
repositories {
google()
jcenter()
}
}

202
android/gradle/wrapper/LICENSE-for-gradlewrapper.txt vendored Normal file
View File

@ -0,0 +1,202 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "{}"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright {yyyy} {name of copyright owner}
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

BIN
android/gradle/wrapper/gradle-wrapper.jar vendored Normal file
View File

Binary file not shown.

1
android/gradle/wrapper/gradle-wrapper.properties vendored Normal file
View File

@ -0,0 +1 @@
distributionUrl=https\://services.gradle.org/distributions/gradle-4.4-all.zip

160
android/gradlew vendored Executable file
View File

@ -0,0 +1,160 @@
#!/usr/bin/env bash
##############################################################################
##
## Gradle start up script for UN*X
##
##############################################################################
# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
DEFAULT_JVM_OPTS=""
APP_NAME="Gradle"
APP_BASE_NAME=`basename "$0"`
# Use the maximum available, or set MAX_FD != -1 to use that value.
MAX_FD="maximum"
warn ( ) {
echo "$*"
}
die ( ) {
echo
echo "$*"
echo
exit 1
}
# OS specific support (must be 'true' or 'false').
cygwin=false
msys=false
darwin=false
case "`uname`" in
CYGWIN* )
cygwin=true
;;
Darwin* )
darwin=true
;;
MINGW* )
msys=true
;;
esac
# Attempt to set APP_HOME
# Resolve links: $0 may be a link
PRG="$0"
# Need this for relative symlinks.
while [ -h "$PRG" ] ; do
ls=`ls -ld "$PRG"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
PRG="$link"
else
PRG=`dirname "$PRG"`"/$link"
fi
done
SAVED="`pwd`"
cd "`dirname \"$PRG\"`/" >/dev/null
APP_HOME="`pwd -P`"
cd "$SAVED" >/dev/null
CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
# Determine the Java command to use to start the JVM.
if [ -n "$JAVA_HOME" ] ; then
if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
# IBM's JDK on AIX uses strange locations for the executables
JAVACMD="$JAVA_HOME/jre/sh/java"
else
JAVACMD="$JAVA_HOME/bin/java"
fi
if [ ! -x "$JAVACMD" ] ; then
die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
Please set the JAVA_HOME variable in your environment to match the
location of your Java installation."
fi
else
JAVACMD="java"
which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
Please set the JAVA_HOME variable in your environment to match the
location of your Java installation."
fi
# Increase the maximum file descriptors if we can.
if [ "$cygwin" = "false" -a "$darwin" = "false" ] ; then
MAX_FD_LIMIT=`ulimit -H -n`
if [ $? -eq 0 ] ; then
if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then
MAX_FD="$MAX_FD_LIMIT"
fi
ulimit -n $MAX_FD
if [ $? -ne 0 ] ; then
warn "Could not set maximum file descriptor limit: $MAX_FD"
fi
else
warn "Could not query maximum file descriptor limit: $MAX_FD_LIMIT"
fi
fi
# For Darwin, add options to specify how the application appears in the dock
if $darwin; then
GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\""
fi
# For Cygwin, switch paths to Windows format before running java
if $cygwin ; then
APP_HOME=`cygpath --path --mixed "$APP_HOME"`
CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
JAVACMD=`cygpath --unix "$JAVACMD"`
# We build the pattern for arguments to be converted via cygpath
ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null`
SEP=""
for dir in $ROOTDIRSRAW ; do
ROOTDIRS="$ROOTDIRS$SEP$dir"
SEP="|"
done
OURCYGPATTERN="(^($ROOTDIRS))"
# Add a user-defined pattern to the cygpath arguments
if [ "$GRADLE_CYGPATTERN" != "" ] ; then
OURCYGPATTERN="$OURCYGPATTERN|($GRADLE_CYGPATTERN)"
fi
# Now convert the arguments - kludge to limit ourselves to /bin/sh
i=0
for arg in "$@" ; do
CHECK=`echo "$arg"|egrep -c "$OURCYGPATTERN" -`
CHECK2=`echo "$arg"|egrep -c "^-"` ### Determine if an option
if [ $CHECK -ne 0 ] && [ $CHECK2 -eq 0 ] ; then ### Added a condition
eval `echo args$i`=`cygpath --path --ignore --mixed "$arg"`
else
eval `echo args$i`="\"$arg\""
fi
i=$((i+1))
done
case $i in
(0) set -- ;;
(1) set -- "$args0" ;;
(2) set -- "$args0" "$args1" ;;
(3) set -- "$args0" "$args1" "$args2" ;;
(4) set -- "$args0" "$args1" "$args2" "$args3" ;;
(5) set -- "$args0" "$args1" "$args2" "$args3" "$args4" ;;
(6) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" ;;
(7) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" ;;
(8) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" ;;
(9) set -- "$args0" "$args1" "$args2" "$args3" "$args4" "$args5" "$args6" "$args7" "$args8" ;;
esac
fi
# Split up the JVM_OPTS And GRADLE_OPTS values into an array, following the shell quoting and substitution rules
function splitJvmOpts() {
JVM_OPTS=("$@")
}
eval splitJvmOpts $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS
JVM_OPTS[${#JVM_OPTS[*]}]="-Dorg.gradle.appname=$APP_BASE_NAME"
exec "$JAVACMD" "${JVM_OPTS[@]}" -classpath "$CLASSPATH" org.gradle.wrapper.GradleWrapperMain "$@"

90
android/gradlew.bat vendored Normal file
View File

@ -0,0 +1,90 @@
@if "%DEBUG%" == "" @echo off
@rem ##########################################################################
@rem
@rem Gradle startup script for Windows
@rem
@rem ##########################################################################
@rem Set local scope for the variables with windows NT shell
if "%OS%"=="Windows_NT" setlocal
@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
set DEFAULT_JVM_OPTS=
set DIRNAME=%~dp0
if "%DIRNAME%" == "" set DIRNAME=.
set APP_BASE_NAME=%~n0
set APP_HOME=%DIRNAME%
@rem Find java.exe
if defined JAVA_HOME goto findJavaFromJavaHome
set JAVA_EXE=java.exe
%JAVA_EXE% -version >NUL 2>&1
if "%ERRORLEVEL%" == "0" goto init
echo.
echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
echo.
echo Please set the JAVA_HOME variable in your environment to match the
echo location of your Java installation.
goto fail
:findJavaFromJavaHome
set JAVA_HOME=%JAVA_HOME:"=%
set JAVA_EXE=%JAVA_HOME%/bin/java.exe
if exist "%JAVA_EXE%" goto init
echo.
echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
echo.
echo Please set the JAVA_HOME variable in your environment to match the
echo location of your Java installation.
goto fail
:init
@rem Get command-line arguments, handling Windowz variants
if not "%OS%" == "Windows_NT" goto win9xME_args
if "%@eval[2+2]" == "4" goto 4NT_args
:win9xME_args
@rem Slurp the command line arguments.
set CMD_LINE_ARGS=
set _SKIP=2
:win9xME_args_slurp
if "x%~1" == "x" goto execute
set CMD_LINE_ARGS=%*
goto execute
:4NT_args
@rem Get arguments from the 4NT Shell from JP Software
set CMD_LINE_ARGS=%$
:execute
@rem Setup the command line
set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
@rem Execute Gradle
"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %CMD_LINE_ARGS%
:end
@rem End local scope for the variables with windows NT shell
if "%ERRORLEVEL%"=="0" goto mainEnd
:fail
rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
rem the _cmd.exe /c_ return code!
if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1
exit /b 1
:mainEnd
if "%OS%"=="Windows_NT" endlocal
:omega

1
android/settings.gradle Normal file
View File

@ -0,0 +1 @@
include ':lib'

124
examples/normStreamer.cpp
View File

@ -71,7 +71,16 @@ class NormStreamer
loopback = state;
if (NORM_SESSION_INVALID != norm_session)
NormSetMulticastLoopback(norm_session, state);
}
}
void SetFtiInfo(bool state)
{
fti_info = state;
if (NORM_SESSION_INVALID != norm_session)
NormLimitObjectInfo(norm_session, state);
}
void SetAckEx(bool state)
{ack_ex = state;}
bool EnableUdpRelay(const char* relayAddr, unsigned short relayPort);
bool EnableUdpListener(unsigned short thePort, const char* groupAddr, const char * interfaceName);
@ -302,6 +311,8 @@ class NormStreamer
bool norm_acking;
bool tx_ack_pending;
NormFlushMode flush_mode; // TBD - allow for "none", "passive", "active" options
bool fti_info;
bool ack_ex;
// Receive stream and state variables for writing received messages to output
NormObjectHandle rx_stream;
@ -324,7 +335,7 @@ class NormStreamer
// These are some options mainly for testing purposes
bool omit_header; // if "true", receive message length header is _not_ written to output
//bool rx_silent;
bool rx_silent;
//double tx_loss;
unsigned long input_byte_count;
unsigned long tx_byte_count;
@ -351,11 +362,11 @@ NormStreamer::NormStreamer()
tx_stream (NORM_OBJECT_INVALID), tx_ready(true),
tx_stream_buffer_max(0), tx_stream_buffer_count(0), tx_stream_bytes_remain(0),
tx_watermark_pending(false), norm_acking(false), tx_ack_pending(false), flush_mode(NORM_FLUSH_ACTIVE),
rx_stream(NORM_OBJECT_INVALID), rx_ready(false), rx_needed(false), msg_sync(false),
fti_info(false), ack_ex(false), rx_stream(NORM_OBJECT_INVALID), rx_ready(false), rx_needed(false), msg_sync(false),
output_bucket_rate(0.0), output_bucket_interval(0.0), output_bucket_depth(0), output_bucket_count(0),
output_socket(ProtoSocket::UDP), output_file(stdout), output_fd(fileno(stdout)), output_ready(true),
output_msg_length(0), output_index(0),
omit_header(false), input_byte_count(0), tx_byte_count(0),
omit_header(false), rx_silent(false), input_byte_count(0), tx_byte_count(0),
segment_size(1398), block_size(64), num_parity(0), auto_parity(0),
stream_buffer_size(2*1024*1024),
tx_socket_buffer_size(0), rx_socket_buffer_size(0),
@ -626,6 +637,8 @@ bool NormStreamer::Start(bool sender, bool receiver)
gettimeofday(&currentTime, NULL);
srand(currentTime.tv_usec); // seed random number generator
NormSessionId instanceId = (NormSessionId)rand();
if (fti_info)
NormLimitObjectInfo(norm_session, true);
if (!NormStartSender(norm_session, instanceId, bufferSize, segment_size, block_size, num_parity))
{
fprintf(stderr, "normStreamer error: unable to start NORM sender\n");
@ -758,7 +771,15 @@ void NormStreamer::ReadInput()
NormStreamClose(tx_stream, true);
if (norm_acking)
{
NormSetWatermark(norm_session, tx_stream, true);
if (ack_ex)
{
const char* req = "Hello, acker";
NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true);
}
else
{
NormSetWatermark(norm_session, tx_stream, true);
}
tx_ack_pending = false;
}
input_needed = false;
@ -840,7 +861,16 @@ unsigned int NormStreamer::WriteToStream(const char* buffer, unsigned int numByt
{
// Initiate flow control ACK request
//fprintf(stderr, "write-initiated flow control ACK REQUEST\n");
NormSetWatermark(norm_session, tx_stream);
if (ack_ex)
{
const char* req = "Hello, acker";
NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1);
}
else
{
NormSetWatermark(norm_session, tx_stream);
}
tx_watermark_pending = true;
tx_ack_pending = false;
}
@ -915,7 +945,15 @@ void NormStreamer::FlushStream(bool eom, NormFlushMode flushMode)
}
if (setWatermark)
{
NormSetWatermark(norm_session, tx_stream, true);
if (ack_ex)
{
const char* req = "Hello, acker";
NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true);
}
else
{
NormSetWatermark(norm_session, tx_stream, true);
}
tx_ack_pending = false;
}
}
@ -1109,6 +1147,7 @@ void NormStreamer::HandleNormEvent(const NormEvent& event)
break;
case NORM_TX_WATERMARK_COMPLETED:
TRACE("NORM_TX_WATERMARK_COMPLETED ...\n");
if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session))
{
//fprintf(stderr, "WATERMARK COMPLETED\n");
@ -1120,7 +1159,15 @@ void NormStreamer::HandleNormEvent(const NormEvent& event)
//fprintf(stderr, "flow control ACK completed\n");
if (tx_ack_pending)
{
NormSetWatermark(norm_session, tx_stream, true);
if (ack_ex)
{
const char* req = "Hello, acker";
NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true);
}
else
{
NormSetWatermark(norm_session, tx_stream, true);
}
tx_ack_pending = false;
}
}
@ -1137,7 +1184,15 @@ void NormStreamer::HandleNormEvent(const NormEvent& event)
if (tx_ack_pending)
{
// May as well advance the ack request point
NormSetWatermark(norm_session, tx_stream, true);
if (ack_ex)
{
const char* req = "Hello, acker";
NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true);
}
else
{
NormSetWatermark(norm_session, tx_stream, true);
}
tx_ack_pending = false;
}
else
@ -1145,6 +1200,26 @@ void NormStreamer::HandleNormEvent(const NormEvent& event)
NormResetWatermark(norm_session);
}
}
if (ack_ex)
{
// This iterates through the acking nodes looking for responses
// to our application-defined NormSetWatermarkEx() request
NormAckingStatus ackingStatus;
NormNodeId nodeId = NORM_NODE_NONE; // this inits NormGetNextAckingNode() iteration
while (NormGetNextAckingNode(event.session, &nodeId, &ackingStatus))
{
if (NORM_ACK_SUCCESS == ackingStatus)
{
// This node acked, so look for AckEx response
// In our example/test case here, we use strings for the content
char buffer[256];
buffer[0] = '\0';
unsigned int buflen = 256;
if (NormGetAckEx(event.session, nodeId, buffer, &buflen))
fprintf(stderr, "Received APP_ACK from node>%u \"%s\"\n", nodeId, buffer);
}
}
}
break;
case NORM_TX_OBJECT_PURGED:
@ -1209,6 +1284,19 @@ void NormStreamer::HandleNormEvent(const NormEvent& event)
rx_needed = false;
break;
case NORM_RX_ACK_REQUEST:
{
char buffer[256];
buffer[0] = '\0';
unsigned int buflen = 256;
NormNodeGetWatermarkEx(event.sender, buffer, &buflen);
fprintf(stderr, "Received NORM_RX_ACK_REQUEST: \"%s\"\n", buffer);
// Send a reply
const char* ack = "Yes, master";
NormNodeSendAckEx(event.sender, ack, strlen(ack) + 1);
break;
}
default:
break;
}
@ -1218,7 +1306,7 @@ void NormStreamer::HandleNormEvent(const NormEvent& event)
void Usage()
{
fprintf(stderr, "Usage: normStreamer id <nodeId> {send|recv} [addr <addr>[/<port>]][interface <name>][loopback]\n"
fprintf(stderr, "Usage: normStreamer id <nodeId> {send|recv} [addr <addr>[/<port>]][interface <name>][loopback][info]\n"
" [cc|cce|ccl|rate <bitsPerSecond>][ack <node1>[,<node2>,...][flush {none|passive|active}]\n"
" [listen [<mcastAddr>/]<port>][linterface <name>]\n"
" [relay <dstAddr>/<port>][limit [<rate>/]<depth>][output <device>]\n"
@ -1244,6 +1332,7 @@ void PrintHelp()
" interface <name> -- Specifies the name of the network interface on which to conduct NORM protocol\n"
" (e.g., 'eth0')\n"
" loopback -- Enables 'loopback' sessions on the same host machine. Required for multicast loopback.\n"
" info -- Limits FTI header extension to NORM_INFO message only (reduced overhead)\n"
" rate <bitsPerSecond> -- sets fixed sender rate (and receiver token bucket rate if 'limit' option is used)\n"
" [cc|cce|ccl] -- Enables optional NORM congestion control mode (overrides 'rate')\n"
" ack [<nodeId list>] -- Instructs sender to request positive acknowledgement from listed receiver nodes\n"
@ -1301,6 +1390,8 @@ int main(int argc, char* argv[])
unsigned int ackingNodeCount = 0;
bool loopback = false;
bool ftiInfo = false;
bool ackEx = false;
int debugLevel = 0;
bool trace = false;
const char* logFile = NULL;
@ -1343,6 +1434,15 @@ int main(int argc, char* argv[])
{
loopback = true;
}
else if (0 == strncmp(cmd, "info", len))
{
ftiInfo = true;
}
else if (0 == strncmp(cmd, "ex", len))
{
// This enables testing/demonstrating NormSetWatermarkEx() operation
ackEx = true;
}
else if (0 == strncmp(cmd, "addr", len))
{
if (i >= argc)
@ -1881,6 +1981,8 @@ int main(int argc, char* argv[])
normStreamer.SetStreamBufferSize(streamBufferSize);
normStreamer.SetLoopback(loopback);
normStreamer.SetFtiInfo(ftiInfo);
normStreamer.SetAckEx(ackEx);
if (omitHeaderOnOutput) normStreamer.OmitHeader(true);
@ -2129,7 +2231,7 @@ int main(int argc, char* argv[])
}
// for debugging to see if anything gets "stuck"
if ((thisTime.tv_sec - lastTime.tv_sec) >= 1)
if ((thisTime.tv_sec - lastTime.tv_sec) >= 100)
{
if (send)
fprintf(stderr, "normStreamer: inputNeeded:%d inputReady:%d txPending:%d txReady:%d inputCount:%lu txCount:%lu fdMask:%d\n",

36
include/normApi.h
View File

@ -188,6 +188,7 @@ typedef enum NormEventType
NORM_RX_OBJECT_UPDATED,
NORM_RX_OBJECT_COMPLETED,
NORM_RX_OBJECT_ABORTED,
NORM_RX_ACK_REQUEST, // upon receipt of app-extended watermark ack request
NORM_GRTT_UPDATED,
NORM_CC_ACTIVE,
NORM_CC_INACTIVE,
@ -328,6 +329,16 @@ void NormSetTxOnly(NormSessionHandle sessionHandle,
bool txOnly,
bool connectToSessionAddress DEFAULT(false));
NORM_API_LINKAGE
void NormLimitObjectInfo(NormSessionHandle sessionHandle, bool state); // if true, FEC OTI in NORM_INFO only
NORM_API_LINKAGE
bool NormPresetObjectInfo(NormSessionHandle sessionHandle, // FEC OTI is preset and not sent
unsigned long objectSize, // (most useful for NORM_OBJECT_STREAM)
UINT16 segmentSize,
UINT16 numData,
UINT16 numParity);
NORM_API_LINKAGE
void NormSetId(NormSessionHandle sessionHandle, NormNodeId normId);
@ -576,6 +587,15 @@ NORM_API_LINKAGE
bool NormSetWatermark(NormSessionHandle sessionHandle,
NormObjectHandle objectHandle,
bool overrideFlush DEFAULT(false));
NORM_API_LINKAGE
bool NormSetWatermarkEx(NormSessionHandle sessionHandle,
NormObjectHandle objectHandle,
const char* buffer,
unsigned int numBytes,
bool overrideFlush DEFAULT(false));
NORM_API_LINKAGE
bool NormResetWatermark(NormSessionHandle sessionHandle);
@ -607,6 +627,12 @@ bool NormGetNextAckingNode(NormSessionHandle sessionHandle,
NormNodeId* nodeId,
NormAckingStatus* ackingStatus DEFAULT(0));
NORM_API_LINKAGE
bool NormGetAckEx(NormSessionHandle sessionHandle,
NormNodeId nodeId,
char* buffer,
unsigned int* buflen);
NORM_API_LINKAGE
bool NormSendCommand(NormSessionHandle sessionHandle,
const char* cmdBuffer,
@ -784,6 +810,16 @@ bool NormNodeGetCommand(NormNodeHandle remoteSender,
char* buffer,
unsigned int* buflen);
NORM_API_LINKAGE
bool NormNodeSendAckEx(NormNodeHandle remoteSender,
const char* buffer,
unsigned int numBytes);
NORM_API_LINKAGE
bool NormNodeGetWatermarkEx(NormNodeHandle remoteSender,
char* buffer,
unsigned int* buflen);
NORM_API_LINKAGE
void NormNodeFreeBuffers(NormNodeHandle remoteSender);

132
include/normMessage.h
View File

@ -183,7 +183,7 @@ const NormNodeId NORM_NODE_ANY = 0xffffffff;
class NormObjectId
{
public:
NormObjectId() {};
NormObjectId() : value(0) {};
NormObjectId(UINT16 id) {value = id;}
NormObjectId(const NormObjectId& id) {value = id.value;}
operator UINT16() const {return value;}
@ -331,14 +331,16 @@ class NormHeaderExtension
INVALID = 0,
FTI = 64, // FEC Object Transmission Information (FTI) extension
CC_FEEDBACK = 3, // NORM-CC Feedback extension
CC_RATE = 128 // NORM-CC Rate extension
CC_RATE = 128, // NORM-CC Rate extension
APP_ACK = 65 // app-defined ACK extension (see NormSetWatermarkEx())
};
NormHeaderExtension();
virtual ~NormHeaderExtension() {}
virtual void Init(UINT32* theBuffer)
virtual void Init(UINT32* theBuffer, UINT16 numBytes)
{
buffer = theBuffer;
// TBD - should we confirm that 'numBytes' is sufficient
AttachBuffer(theBuffer, numBytes);
SetType(INVALID);
SetWords(0);
}
@ -347,13 +349,17 @@ class NormHeaderExtension
void SetWords(UINT8 words)
{((UINT8*)buffer)[LENGTH_OFFSET] = words;}
void AttachBuffer(const UINT32* theBuffer) {buffer = (UINT32*)theBuffer;}
const UINT32* GetBuffer() {return buffer;}
void AttachBuffer(const UINT32* theBuffer, UINT16 bufferLength)
{
buffer = (UINT32*)theBuffer;
buffer_length = bufferLength;
}
const UINT32* GetBuffer()
{return buffer;}
Type GetType() const
{
return buffer ? (Type)(((UINT8*)buffer)[TYPE_OFFSET]) : INVALID;
}
{return buffer ? (Type)(((UINT8*)buffer)[TYPE_OFFSET]) : INVALID;}
UINT16 GetLength() const
{
return (buffer ?
@ -361,14 +367,27 @@ class NormHeaderExtension
((((UINT8*)buffer)[LENGTH_OFFSET]) << 2) : 4) :
0);
}
// These currently only used for APP_ACK extension
const char* GetContent()
{return (((const char*)buffer) + CONTENT_OFFSET);}
UINT16 GetContentLength() const
{
UINT16 totalLen = GetLength();
return ((totalLen > CONTENT_OFFSET) ? (totalLen - CONTENT_OFFSET) : 0);
}
static UINT16 GetContentOffset()
{return (UINT16)CONTENT_OFFSET;}
protected:
enum
{
TYPE_OFFSET = 0, // UINT8 offset
LENGTH_OFFSET = TYPE_OFFSET + 1 // UINT8 offset
TYPE_OFFSET = 0, // UINT8 offset
LENGTH_OFFSET = TYPE_OFFSET + 1, // UINT8 offset
CONTENT_OFFSET = LENGTH_OFFSET + 1 // UINT8 offset
};
UINT32* buffer;
UINT16 buffer_length;
}; // end class NormHeaderExtension
@ -594,9 +613,15 @@ class NormMsg
void AttachExtension(NormHeaderExtension& extension)
{
extension.Init(buffer+(header_length/4));
extension.Init(buffer+(header_length/4), MAX_SIZE - header_length);
ExtendHeaderLength(extension.GetLength());
}
// Only use this for extensions that have content appended after attachment
// (Fixed-length extensions should set their length upon Init())
void PackExtension(NormHeaderExtension& extension)
{
ExtendHeaderLength(2 + extension.GetContentLength());
}
// Message processing routines
bool InitFromBuffer(UINT16 msgLength);
@ -610,7 +635,7 @@ class NormMsg
{return (((UINT8*)buffer)[VERSION_OFFSET] >> 4);}
NormMsg::Type GetType() const
{return (Type)(((UINT8*)buffer)[TYPE_OFFSET] & 0x0f);}
UINT16 GetHeaderLength()
UINT16 GetHeaderLength() const
{return ((UINT8*)buffer)[HDR_LEN_OFFSET] << 2;}
UINT16 GetBaseHeaderLength()
{return header_length_base;}
@ -633,12 +658,21 @@ class NormMsg
bool HasExtensions() const {return (header_length > header_length_base);}
bool GetNextExtension(NormHeaderExtension& ext) const
{
const UINT32* currentBuffer = ext.GetBuffer();
const UINT32* currentBuffer = ext.GetBuffer();
// 'nextOffset' here is a UINT32 offset
UINT16 nextOffset =
(UINT16)(currentBuffer ? (currentBuffer - buffer + (ext.GetLength()/4)) :
(header_length_base/4));
(header_length_base/4));
bool result = HasExtensions() ? (nextOffset < (header_length/4)) : false;
ext.AttachBuffer(result ? (buffer+nextOffset) : (UINT32*)NULL);
if (result)
{
UINT16 nextLength = ((UINT8*)(buffer + nextOffset))[1] << 2;
ext.AttachBuffer(buffer+nextOffset, nextLength);
}
else
{
ext.AttachBuffer((UINT32*)NULL, 0);
}
return result;
}
@ -753,9 +787,9 @@ class NormFtiExtension2 : public NormHeaderExtension
{
public:
// To build the FTI Header Extension
virtual void Init(UINT32* theBuffer)
virtual void Init(UINT32* theBuffer, UINT16 numBytes)
{
AttachBuffer(theBuffer);
AttachBuffer(theBuffer, numBytes);
SetType(FTI); // HET = 64
SetWords(4);
}
@ -814,6 +848,16 @@ class NormFtiData
num_data(0), num_parity(0), fec_m(0), instance_id(0) {}
~NormFtiData() {}
bool IsValid() const
{return (0 != segment_size);}
void Invalidate()
{
object_size = 0;
segment_size = num_data = num_parity = instance_id = 0;
fec_m = 0;
}
void SetObjectSize(const NormObjectSize& objectSize)
{object_size = objectSize;}
void SetSegmentSize(UINT16 segmentSize)
@ -850,16 +894,15 @@ class NormFtiData
}; // end class NormFtiData
// This FEC Object Transmission Information assumes "fec_id" == 5 (RFC 5510)
// (this is the fully-defined 8-bit Reed-Solomon codec)
class NormFtiExtension5 : public NormHeaderExtension
{
public:
// To build the fec_id=5 FTI Header Extension
virtual void Init(UINT32* theBuffer)
virtual void Init(UINT32* theBuffer, UINT16 numBytes)
{
AttachBuffer(theBuffer);
AttachBuffer(theBuffer, numBytes);
SetType(FTI); // HET = 64
SetWords(3);
}
@ -891,7 +934,7 @@ class NormFtiExtension5 : public NormHeaderExtension
private:
enum
{
OBJ_SIZE_MSB_OFFSET = (LENGTH_OFFSET + 1)/2,
OBJ_SIZE_MSB_OFFSET = (CONTENT_OFFSET)/2, // UINT16 offset
OBJ_SIZE_LSB_OFFSET = ((OBJ_SIZE_MSB_OFFSET*2)+2)/4,
SEG_SIZE_OFFSET = ((OBJ_SIZE_LSB_OFFSET*4)+4)/2,
FEC_NDATA_OFFSET = ((SEG_SIZE_OFFSET+1)*2),
@ -899,6 +942,37 @@ class NormFtiExtension5 : public NormHeaderExtension
};
}; // end class NormFtiExtension5
class NormAppAckExtension : public NormHeaderExtension
{
public:
virtual void Init(UINT32* theBuffer, UINT16 numBytes)
{
AttachBuffer(theBuffer, numBytes);
SetType(APP_ACK);
SetWords(0);
}
bool SetContent(const char* data, UINT16 dataLen)
{
if (dataLen > (buffer_length - CONTENT_OFFSET)) return false;
memcpy(((char*)buffer) + CONTENT_OFFSET, data, dataLen);
if (dataLen > 2)
{
// Pad out to get 32-bit alignment of extension
dataLen += CONTENT_OFFSET;
UINT16 padLen = dataLen % 4;
if (padLen)
padLen = 4 - padLen;
dataLen += padLen;
SetWords(dataLen/4);
}
else
{
SetWords(1);
}
return true;
}
}; // end class NormAppAckExtension
// This FEC Object Transmission Information assumes "fec_id" == 129
class NormFtiExtension129 : public NormHeaderExtension
@ -906,9 +980,9 @@ class NormFtiExtension129 : public NormHeaderExtension
public:
// To build the FTI Header Extension
// (TBD) allow for different "fec_id" types in the future
virtual void Init(UINT32* theBuffer)
virtual void Init(UINT32* theBuffer, UINT16 numBytes)
{
AttachBuffer(theBuffer);
AttachBuffer(theBuffer, numBytes);
SetType(FTI);
SetWords(4);
}
@ -1204,7 +1278,7 @@ class NormCmdFlushMsg : public NormCmdMsg
{length = header_length;}
bool AppendAckingNode(NormNodeId nodeId, UINT16 segmentSize)
{
if ((length-header_length+ 4) > segmentSize) return false;
if ((length-header_length + 4) > segmentSize) return false;
buffer[length/4] = htonl((UINT32)nodeId);
length += 4;
return true;
@ -1448,9 +1522,9 @@ class NormCCRateExtension : public NormHeaderExtension
{
public:
virtual void Init(UINT32* theBuffer)
virtual void Init(UINT32* theBuffer, UINT16 numBytes)
{
AttachBuffer(theBuffer);
AttachBuffer(theBuffer, numBytes);
SetType(CC_RATE);
((UINT8*)buffer)[RESERVED_OFFSET] = 0;
}
@ -1668,9 +1742,9 @@ class NormCmdRepairAdvMsg : public NormCmdMsg
class NormCCFeedbackExtension : public NormHeaderExtension
{
public:
virtual void Init(UINT32* theBuffer)
virtual void Init(UINT32* theBuffer, UINT16 numBytes)
{
AttachBuffer(theBuffer);
AttachBuffer(theBuffer, numBytes);
SetType(CC_FEEDBACK);
SetWords(3);
((UINT8*)buffer)[CC_FLAGS_OFFSET] = 0;

51
include/normNode.h
View File

@ -206,10 +206,21 @@ class NormAckingNode : public NormNode
unsigned int GetReqCount() const {return req_count;}
bool AckReceived() const {return ack_received;}
void MarkAckReceived() {ack_received = true;}
bool SetAckEx(const char* buffer, UINT16 numBytes);
bool GetAckEx(char* buffer, unsigned int* buflen);
/*
const char* GetAppAckContent() const
{return (const char*)ack_ex_buffer;}
unsigned int GetAppAckLength() const
{return ack_ex_length;} */
private:
bool ack_received; // was ack received?
unsigned int req_count; // remaining request attempts
char* ack_ex_buffer;
unsigned int ack_ex_length;
}; // end NormAckingNode
@ -297,7 +308,7 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
UINT16 segmentSize ,
UINT16 numData,
UINT16 numParity);
bool GetFtiData(const NormObjectMsg& msg, NormFtiData& ftiData);
bool GetFtiData(const NormObjectMsg& msg, NormFtiData& ftiData);
// Parameters
NormObject::NackingMode GetDefaultNackingMode() const
@ -359,7 +370,7 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
UINT16 segmentSize,
UINT16 numData,
UINT16 numParity);
bool BuffersAllocated() {return (0 != segment_size);}
bool BuffersAllocated() {return (NULL != retrieval_pool);}
void FreeBuffers();
void Activate(bool isObjectMsg);
@ -370,7 +381,7 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
UINT8 GetFecFieldSize() const
{return fec_m;}
{return fti_data.GetFecFieldSize();}
bool GetFirstPending(NormObjectId& objectId) const
{
@ -408,9 +419,9 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
return NULL;
}
UINT16 SegmentSize() {return segment_size;}
UINT16 BlockSize() {return ndata;}
UINT16 NumParity() {return nparity;}
UINT16 SegmentSize() const {return fti_data.GetSegmentSize();}
UINT16 BlockSize() const {return fti_data.GetFecMaxBlockLen();}
UINT16 NumParity() const {return fti_data.GetFecNumParity();}
NormBlock* GetFreeBlock(NormObjectId objectId, NormBlockId blockId);
void PutFreeBlock(NormBlock* block)
@ -425,7 +436,7 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
void SetErasureLoc(UINT16 index, UINT16 value)
{
ASSERT(index < nparity);
ASSERT(index < NumParity());
erasure_loc[index] = value;
}
UINT16 GetErasureLoc(UINT16 index)
@ -434,7 +445,7 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
}
void SetRetrievalLoc(UINT16 index, UINT16 value)
{
ASSERT(index < ndata);
ASSERT(index < BlockSize());
retrieval_loc[index] = value;
}
UINT16 GetRetrievalLoc(UINT16 index)
@ -444,7 +455,7 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
char* GetRetrievalSegment()
{
char* s = retrieval_pool[retrieval_index++];
retrieval_index = (retrieval_index >= ndata) ? 0 : retrieval_index;
retrieval_index = (retrieval_index >= BlockSize()) ? 0 : retrieval_index;
return s;
}
@ -458,9 +469,9 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
// Statistics kept on sender
unsigned long CurrentBufferUsage() const
{return (segment_size * segment_pool.CurrentUsage());}
{return (SegmentSize() * segment_pool.CurrentUsage());}
unsigned long PeakBufferUsage() const
{return (segment_size * segment_pool.PeakUsage());}
{return (SegmentSize() * segment_pool.PeakUsage());}
unsigned long BufferOverunCount() const
{return segment_pool.OverunCount() + block_pool.OverrunCount();}
@ -530,6 +541,9 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
bool ReadNextCmd(char* buffer, unsigned int* buflen);
bool SendAckEx(const char* data, unsigned int numBytes);
bool GetWatermarkEx(char* buffer, unsigned int* buflen);
void SetAddress(const ProtoAddress& address)
{
unsigned int len = address.GetLength();
@ -591,11 +605,15 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
UINT16 max_pending_range; // max range of pending objs allowed
bool is_open;
UINT16 segment_size;
// TBD - embed the FTI parameters into a NormFtiData object
UINT8 fec_id;
UINT8 fec_m;
unsigned int ndata;
unsigned int nparity;
NormFtiData fti_data;
bool preset_fti;
//UINT16 segment_size;
//UINT8 fec_id;
//UINT8 fec_m;
//unsigned int ndata;
//unsigned int nparity;
NormStreamObject* preset_stream;
NormObjectTable rx_table;
@ -622,6 +640,9 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
NormObjectId watermark_object_id;
NormBlockId watermark_block_id;
NormSegmentId watermark_segment_id;
bool ack_ex_pending;
char* ack_ex_buffer;
unsigned int ack_ex_length;
// Remote sender grtt measurement state
double grtt_estimate;

9
include/normObject.h
View File

@ -89,7 +89,14 @@ class NormObject
info_len = 0;
pending_info = false;
}
// This is used to bootstrap reception sender
// is using FTI_INFO FtiMode
void SetPendingInfo(bool state, UINT8 fecId)
{
pending_info = true;
fec_id = fecId;
}
const char* GetInfo() const {return info_ptr;}
UINT16 GetInfoLength() const {return info_len;}
bool IsStream() const {return (STREAM == type);}

45
include/normSession.h
View File

@ -51,6 +51,7 @@ class NormController
RX_OBJECT_UPDATED,
RX_OBJECT_COMPLETED,
RX_OBJECT_ABORTED,
RX_ACK_REQUEST, // upon receipt of app-extended watermark ack request
GRTT_UPDATED,
CC_ACTIVE, // posted when cc feedback is detected
CC_INACTIVE, // posted when no cc feedback and min rate reached
@ -166,6 +167,14 @@ class NormSession
TRACK_SENDERS = 0x02,
TRACK_ALL = 0x03
};
// Object FEC Transport Information (FTI) mode
enum FtiMode
{
FTI_PRESET = 0, // Receivers have preset FTI, don't send
FTI_INFO = 1, // Send FTI in NORM_INFO messages only
FTI_ALWAYS = 2 // Send FTI in NORM_DATA and NORM_INFO messages
};
// General methods
void SetNodeId(NormNodeId nodeId)
@ -384,11 +393,14 @@ class NormSession
NormObject* SenderFindTxObject(NormObjectId objectId)
{return tx_table.Find(objectId);}
// postive ack mgmnt
void SenderSetWatermark(NormObjectId objectId,
// postive ack mgmnt (can only fail when 'appAckReq' is set)
bool SenderSetWatermark(NormObjectId objectId,
NormBlockId blockId,
NormSegmentId segmentId,
bool overrideFlush = false);
bool overrideFlush = false,
const char* appAckReq = NULL,
unsigned int appAckReqLen = 0);
void SenderResetWatermark();
void SenderCancelWatermark();
@ -400,6 +412,7 @@ class NormSession
AckingStatus SenderGetAckingStatus(NormNodeId nodeId);
// Set "prevNodeId = NORM_NODE_NONE" to init this iteration (returns "false" when done)
bool SenderGetNextAckingNode(NormNodeId& prevNodeId, AckingStatus* ackingStatus = NULL);
bool SenderGetAckEx(NormNodeId nodeId, char* buffer, unsigned int* buflen);
NormAckingNode* SenderFindAckingNode(NormNodeId nodeId) const
{
@ -486,6 +499,11 @@ class NormSession
gsize_advertised = NormUnquantizeGroupSize(gsize_quantized);
}
FtiMode SenderFtiMode() const
{return fti_mode;}
void SenderSetFtiMode(FtiMode ftiMode)
{fti_mode = ftiMode;}
void SenderEncode(unsigned int segmentId, const char* segment, char** parityVectorList)
{encoder->Encode(segmentId, segment, parityVectorList);}
@ -530,6 +548,21 @@ class NormSession
UINT16 numParity,
unsigned int streamBufferSize = 0);
bool SetPresetFtiData(unsigned int objectSize,
UINT16 segmentSize,
UINT16 numData,
UINT16 numParity);
bool GetPresetFtiData(NormFtiData& ftiData)
{
if (preset_fti.IsValid())
{
ftiData = preset_fti;
return true;
}
return false;
}
void ReceiverSetUnicastNacks(bool state)
{unicast_nacks = state;}
bool ReceiverGetUnicastNacks() const
@ -724,6 +757,7 @@ class NormSession
bool sndr_emcon;
bool tx_only;
bool tx_connect;
FtiMode fti_mode;
NormObjectTable tx_table;
ProtoSlidingMask tx_pending_mask;
@ -813,6 +847,10 @@ class NormSession
ProtoTimer cmd_timer;
bool syn_status;
// Sender "app-defined" ACK_REQUEST state (for NormSetWatermarkEx())
char* ack_ex_buffer;
unsigned int ack_ex_length;
// Receiver parameters
bool is_receiver;
int rx_robust_factor;
@ -828,6 +866,7 @@ class NormSession
NormObject::NackingMode default_nacking_mode;
NormSenderNode::SyncPolicy default_sync_policy;
UINT16 rx_cache_count_max;
NormFtiData preset_fti;
// For NormSocket server-listener support
bool is_server_listener;

2
include/normVersion.h
View File

@ -36,6 +36,6 @@
#ifndef _NORM_VERSION
#define _NORM_VERSION
#define VERSION "1.5.8"
#define VERSION "1.5.9"
#endif // _NORM_VERSION

2
makefiles/Makefile.common
View File

@ -14,7 +14,7 @@ NS = ../src/sim/ns
INCLUDES = $(SYSTEM_INCLUDES) -I$(UNIX) -I../include -I$(PROTOLIB)/include
CFLAGS = -g -DPROTO_DEBUG -DUNIX -D_FILE_OFFSET_BITS=64 -O $(SYSTEM_CFLAGS) $(SYSTEM_HAVES) $(INCLUDES)
CFLAGS = -g -DPROTO_DEBUG -DUNIX -D_FILE_OFFSET_BITS=64 -O $(SYSTEM_CFLAGS) $(SYSTEM_HAVES) $(INCLUDES) -Wno-attributes
#CFLAGS = -g -DPROTO_DEBUG -DUNIX -D_FILE_OFFSET_BITS=64 $(SYSTEM_CFLAGS) $(SYSTEM_HAVES) $(INCLUDES)
LDFLAGS = $(SYSTEM_LDFLAGS)

0
makefiles/android/AndroidManifest.xml → makefiles/android-ndk/AndroidManifest.xml
View File

8
makefiles/android/README → makefiles/android-ndk/README
View File

@ -1,3 +1,11 @@
IMPORTANT NOTE: This "norm/makefiles/android-ndk" directory contains build
files for the ndk-build command more commonly used in older Android SDK
installations. The newer "norm/android" directory has gradle/CMake files
that are compatible with Android Studio or for command-line builds using the
provided "./gradlew" script.
This directory contains an Android NDK import module for NORM, which builds
a shared library, and the JNI bindings.

0
makefiles/android/ant.properties → makefiles/android-ndk/ant.properties
View File

0
makefiles/android/build.xml → makefiles/android-ndk/build.xml
View File

2
makefiles/android/jni/Android.mk → makefiles/android-ndk/jni/Android.mk
View File

@ -60,4 +60,4 @@ LOCAL_SRC_FILES := \
../../../examples/normMsgr.cpp
include $(BUILD_EXECUTABLE)
$(call import-module,protolib/makefiles/android/jni)
$(call import-module,protolib/makefiles/android-ndk/jni)

0
makefiles/android/jni/Application.mk → makefiles/android-ndk/jni/Application.mk
View File

0
makefiles/android/proguard-project.txt → makefiles/android-ndk/proguard-project.txt
View File

0
makefiles/android/project.properties → makefiles/android-ndk/project.properties
View File

11
norm.pc.in Normal file
View File

@ -0,0 +1,11 @@
prefix=@PREFIX@
exec_prefix=${prefix}
libdir=@LIBDIR@
includedir=${prefix}/include
Name: norm
Version: @VERSION@
Description: NACK-Oriented Reliable Multicast (NORM) library
Libs: -L${libdir} -lnorm
Libs.private: @STATIC_LIBS@
Cflags: -I${includedir}

154
src/common/normApi.cpp
View File

@ -1195,7 +1195,47 @@ void NormSetTxOnly(NormSessionHandle sessionHandle,
}
} // end NormSetTxOnly()
// These functions are used internally by the NormSocket API extensiont
NORM_API_LINKAGE
bool NormPresetObjectInfo(NormSessionHandle sessionHandle,
unsigned long objectSize,
UINT16 segmentSize,
UINT16 numData,
UINT16 numParity)
{
bool result = false;
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormSession* session = (NormSession*)sessionHandle;
if (session)
{
result = session->SetPresetFtiData(objectSize, segmentSize, numData, numParity);
if (result) session->SenderSetFtiMode(NormSession::FTI_PRESET);
}
instance->dispatcher.ResumeThread();
}
return result;
} // end NormPresetObjectInfo()
NORM_API_LINKAGE
void NormLimitObjectInfo(NormSessionHandle sessionHandle, bool state)
{
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormSession* session = (NormSession*)sessionHandle;
if (session)
{
if (state)
session->SenderSetFtiMode(NormSession::FTI_INFO);
else
session->SenderSetFtiMode(NormSession::FTI_ALWAYS);
}
instance->dispatcher.ResumeThread();
}
} // end NormLimitObjectInfo()
// These functions are used internally by the NormSocket API extension
NORM_API_LINKAGE
void NormSetId(NormSessionHandle sessionHandle, NormNodeId normId)
@ -2046,6 +2086,49 @@ bool NormSetWatermark(NormSessionHandle sessionHandle,
return result;
} // end NormSetWatermark()
NORM_API_LINKAGE
bool NormSetWatermarkEx(NormSessionHandle sessionHandle,
NormObjectHandle objectHandle,
const char* buffer,
unsigned int numBytes,
bool overrideFlush)
{
bool result = false;
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormSession* session = (NormSession*)sessionHandle;
NormObject* obj = (NormObject*)objectHandle;
if (session && obj)
{
// Purge any existing NORM_TX_WATERMARK_COMPLETED notifications to be safe
instance->PurgeNotifications(sessionHandle, NORM_TX_WATERMARK_COMPLETED);
// (segmentId doesn't matter for non-stream)
if (obj->IsStream())
{
NormStreamObject* stream = static_cast<NormStreamObject*>(obj);
result = session->SenderSetWatermark(stream->GetId(),
stream->FlushBlockId(),
stream->FlushSegmentId(),
overrideFlush,
buffer, numBytes);
}
else
{
NormBlockId blockId = obj->GetFinalBlockId();
NormSegmentId segmentId = obj->GetBlockSize(blockId) - 1;
result = session->SenderSetWatermark(obj->GetId(),
blockId,
segmentId,
overrideFlush,
buffer, numBytes);
}
}
instance->dispatcher.ResumeThread();
}
return result;
} // end NormSetWatermarkEx()
NORM_API_LINKAGE
bool NormResetWatermark(NormSessionHandle sessionHandle)
{
@ -2188,6 +2271,25 @@ bool NormGetNextAckingNode(NormSessionHandle sessionHandle,
}
} // end NormGetNextAckingNode()
NORM_API_LINKAGE
bool NormGetAckEx(NormSessionHandle sessionHandle,
NormNodeId nodeId,
char* buffer,
unsigned int* buflen)
{
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormSession* session = (NormSession*)sessionHandle;
bool result = (NormAckingStatus)session->SenderGetAckEx(nodeId, buffer, buflen);
instance->dispatcher.ResumeThread();
return result;
}
if (NULL != buflen) *buflen = 0;
return false;
} // end NormGetAckEx()
NORM_API_LINKAGE
bool NormSendCommand(NormSessionHandle sessionHandle,
const char* cmdBuffer,
@ -2523,8 +2625,8 @@ UINT16 NormObjectGetInfoLength(NormObjectHandle objectHandle)
NORM_API_LINKAGE
UINT16 NormObjectGetInfo(NormObjectHandle objectHandle,
char* buffer,
UINT16 bufferLen)
char* buffer,
UINT16 bufferLen)
{
UINT16 result = 0;
if (NORM_OBJECT_INVALID != objectHandle)
@ -2816,6 +2918,52 @@ bool NormNodeGetCommand(NormNodeHandle nodeHandle,
return result;
} // end NormNodeGetCommand()
NORM_API_LINKAGE
bool NormNodeSendAckEx(NormNodeHandle nodeHandle,
const char* buffer,
unsigned int numBytes)
{
bool result = false;
if (NORM_NODE_INVALID != nodeHandle)
{
NormInstance* instance = NormInstance::GetInstanceFromNode(nodeHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormNode* node = (NormNode*)nodeHandle;
if (NormNode::SENDER == node->GetType())
{
NormSenderNode* sender = static_cast<NormSenderNode*>(node);
result = sender->SendAckEx(buffer, numBytes);
}
instance->dispatcher.ResumeThread();
}
}
return result;
} // end NormNodeSendAckEx()
NORM_API_LINKAGE
bool NormNodeGetWatermarkEx(NormNodeHandle nodeHandle,
char* buffer,
unsigned int* buflen)
{
bool result = false;
if (NORM_NODE_INVALID != nodeHandle)
{
NormInstance* instance = NormInstance::GetInstanceFromNode(nodeHandle);
if (instance && instance->dispatcher.SuspendThread())
{
NormNode* node = (NormNode*)nodeHandle;
if (NormNode::SENDER == node->GetType())
{
NormSenderNode* sender = static_cast<NormSenderNode*>(node);
result = sender->GetWatermarkEx(buffer, buflen);
}
instance->dispatcher.ResumeThread();
}
}
return result;
} // end NormNodeGetWatermarkEx()
NORM_API_LINKAGE
void NormNodeFreeBuffers(NormNodeHandle nodeHandle)
{

61
src/common/normApp.cpp
View File

@ -67,6 +67,14 @@ class NormApp : public NormController, public ProtoApp
char file_path[PATH_MAX];
NormObjectId object_id;
}; // end class NormApp::FileCacheItem
class Notification : public ProtoList::Item
{
public:
NormController::Event event;
}; // end class NormApp::Notification
class NotificationQueue : public ProtoListTemplate<Notification> {};
Notification* GetNotification(); // get from pool or create as needed
void ShowHelp();
void OnInputReady();
@ -98,6 +106,9 @@ class NormApp : public NormController, public ProtoApp
NormSession* session;
NormStreamObject* tx_stream;
NormStreamObject* rx_stream;
NotificationQueue notify_queue;
NotificationQueue notify_pool;
bool notify_reentry;
// application parameters
FILE* input; // input stream
@ -215,7 +226,7 @@ NormApp::FileCacheItem::~FileCacheItem()
NormApp::NormApp()
: control_pipe(ProtoPipe::MESSAGE), control_remote(false),
session_mgr(GetTimerMgr(), GetSocketNotifier(), &dispatcher),
session(NULL), tx_stream(NULL), rx_stream(NULL), input(NULL), output(NULL),
session(NULL), tx_stream(NULL), rx_stream(NULL), notify_reentry(false), input(NULL), output(NULL),
output_io_buffer(NULL), output_io_bufsize(0),
input_index(0), input_length(0), input_active(false),
push_mode(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE),
@ -257,6 +268,8 @@ NormApp::NormApp()
NormApp::~NormApp()
{
notify_queue.Destroy();
notify_pool.Destroy();
if (address) delete[] address;
if (interface_name) delete[] interface_name;
@ -272,6 +285,20 @@ NormApp::~NormApp()
}
}
NormApp::Notification* NormApp::GetNotification()
{
Notification* notification = notify_pool.RemoveHead();
if (NULL == notification)
{
if (NULL == (notification = new Notification()))
{
PLOG(PL_ERROR, "NormApp::GetNotification() new Notification error: %s\n", GetErrorString());
return NULL;
}
}
return notification;
} // end NormApp::GetNotification()
// NOTE on message flushing mode:
//
// "none" - messages are aggregated to make NORM_DATA
@ -1557,6 +1584,18 @@ void NormApp::Notify(NormController::Event event,
class NormNode* sender,
class NormObject* object)
{
if (notify_reentry)
{
Notification* notification;
if (NULL == (notification = GetNotification()))
{
PLOG(PL_ERROR, "NormApp::Notify() error: unable to create Notification item!\n");
return;
}
notification->event = event;
return;
}
notify_reentry = true;
switch (event)
{
case TX_QUEUE_VACANCY:
@ -1804,7 +1843,7 @@ void NormApp::Notify(NormController::Event event,
// Read the stream when it's updated
ASSERT(NULL != output);
bool reading = true;
bool findMsgSync;
bool seekMsgStart;
while (reading)
{
unsigned int readLength;
@ -1825,17 +1864,17 @@ void NormApp::Notify(NormController::Event event,
ASSERT(output_msg_length >= 2);
readLength = output_msg_length - output_index;
}
findMsgSync = output_msg_sync ? false : true;
seekMsgStart = output_msg_sync ? false : true;
}
else
{
output_index = 0;
readLength = 512;
findMsgSync = false;
seekMsgStart = false;
}
if(!((NormStreamObject*)object)->Read(output_buffer+output_index,
&readLength, findMsgSync))
&readLength, seekMsgStart))
{
// The stream broke
if (output_messaging)
@ -1844,8 +1883,8 @@ void NormApp::Notify(NormController::Event event,
PLOG(PL_ERROR, "NormApp::Notify() detected broken stream ...\n");
output_msg_length = output_index = 0;
output_msg_sync = false;
if (findMsgSync)
break;
if (seekMsgStart)
break; // didn't find msg start yet
else
continue; // try to re-sync
}
@ -2019,6 +2058,14 @@ void NormApp::Notify(NormController::Event event,
break;
}
} // end switch(event)
notify_reentry = false;
// Now handle any deferred, reentrant notifications
Notification* notification;
while (NULL != (notification = notify_queue.RemoveHead()))
{
Notify(notification->event, sessionMgr, session, sender, object);
notify_pool.Append(*notification);
}
} // end NormApp::Notify()

4
src/common/normEncoderRS16.cpp
View File

@ -784,7 +784,7 @@ bool NormDecoderRS16::InvertDecodingMatrix()
icol = col ;
goto found_piv ;
}
for (int row = 0 ; row < k ; row++)
for (unsigned int row = 0 ; row < k ; row++)
{
if (inv_pivt[row] != 1)
{
@ -879,7 +879,7 @@ bool NormDecoderRS16::InvertDecodingMatrix()
}
else if (inv_ndxr[col] != inv_ndxc[col] )
{
for (int row = 0 ; row < k ; row++ )
for (unsigned int row = 0 ; row < k ; row++ )
SWAP( src[row*k + inv_ndxr[col]], src[row*k + inv_ndxc[col]], gf) ;
}
}

2
src/common/normMessage.cpp
View File

@ -49,7 +49,7 @@ bool NormMsg::InitFromBuffer(UINT16 msgLength)
PLOG(PL_FATAL, "NormMsg::InitFromBuffer(DATA) unknown fec_id value: %u\n", fecId);
return false;
}
header_length_base = 4*NormCmdFlushMsg::FEC_ID_OFFSET + fecIdLen;
header_length_base = 4 * NormCmdFlushMsg::FEC_PAYLOAD_ID_OFFSET + fecIdLen;
break;
}
case NormCmdMsg::EOT:

473
src/common/normNode.cpp
View File

@ -9,12 +9,10 @@ NormNode::NormNode(Type nodeType, class NormSession& theSession, NormNodeId node
: session(theSession), node_type(nodeType), id(nodeId), reference_count(1), user_data(NULL),
parent(NULL), right(NULL), left(NULL)
{
}
NormNode::~NormNode()
{
//TRACE("NormNode %lu dtor ...\n", (unsigned long)id);
}
@ -68,9 +66,11 @@ const double NormSenderNode::ACTIVITY_INTERVAL_MIN = 1.0; // 1 second min activ
NormSenderNode::NormSenderNode(class NormSession& theSession, NormNodeId nodeId)
: NormNode(SENDER, theSession, nodeId), instance_id(0), robust_factor(session.GetRxRobustFactor()),
synchronized(false), sync_id(0),
is_open(false), segment_size(0), fec_m(0), ndata(0), nparity(0), preset_stream(NULL),
is_open(false), preset_fti(false), preset_stream(NULL),
repair_boundary(BLOCK_BOUNDARY), decoder(NULL), erasure_loc(NULL),
retrieval_loc(NULL), retrieval_pool(NULL), ack_pending(false), notify_on_grtt_update(true),
retrieval_loc(NULL), retrieval_pool(NULL), ack_pending(false),
ack_ex_pending(false), ack_ex_buffer(NULL), ack_ex_length(0),
notify_on_grtt_update(true),
cc_sequence(0), cc_enable(false), cc_feedback_needed(false), cc_rate(0.0),
rtt_confirmed(false), is_clr(false), is_plr(false),
slow_start(true), send_rate(0.0), recv_rate(0.0), recv_rate_prev(0.0),
@ -164,6 +164,13 @@ void NormSenderNode::Close()
if (ack_timer.IsActive()) ack_timer.Deactivate();
FreeBuffers();
if (NULL != ack_ex_buffer)
{
delete[] ack_ex_buffer;
ack_ex_buffer = NULL;
ack_ex_length = 0;
}
// Delete any command buffers from cmd_buffer queue
while (NULL != cmd_buffer_head)
{
@ -238,8 +245,7 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
return false;
}
// The extra byte of segments is used for marking segments
// which are "start segments" for messages encapsulated in a NormStreamObject
// Segment buffers include space for NORM_OBJECT_STREAM stream payload header
if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()))
{
PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() segment_pool init error\n");
@ -249,7 +255,7 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
// The "retrieval_pool" is used for FEC block decoding
// These segments are temporarily used for "retrieved" source symbol segments
// needed for block decoding (new rx buffer mgmt scheme)
// that aren't still cached and needed for block decoding
if (!(retrieval_pool = new char*[numData]))
{
PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new retrieval_pool error: %s\n", GetErrorString());
@ -366,12 +372,14 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
{
decoder = NULL;
} // end if/else (0 != numParity)
segment_size = segmentSize;
fti_data.SetSegmentSize(segmentSize);
nominal_packet_size = (double)segmentSize;
fec_id = fecId;
fec_m = fecM;
ndata = numData;
nparity = numParity;
fti_data.SetFecFieldSize(fecM);
fti_data.SetFecMaxBlockLen(numData);
fti_data.SetFecNumParity(numParity);
IncrementResyncCount();
return true;
} // end NormSenderNode::AllocateBuffers()
@ -396,7 +404,8 @@ void NormSenderNode::FreeBuffers()
}
if (retrieval_pool)
{
for (unsigned int i = 0; i < ndata; i++)
UINT16 numData = BlockSize();
for (unsigned int i = 0; i < numData; i++)
{
if (retrieval_pool[i])
{
@ -407,7 +416,6 @@ void NormSenderNode::FreeBuffers()
delete[] retrieval_pool;
retrieval_pool = NULL;
}
NormObject* obj;
while ((obj = rx_table.Find(rx_table.RangeLo())))
{
@ -418,7 +426,7 @@ void NormSenderNode::FreeBuffers()
}
segment_pool.Destroy();
block_pool.Destroy();
segment_size = ndata = nparity = 0;
fti_data.Invalidate();
} // end NormSenderNode::FreeBuffers()
unsigned long NormSenderNode::CurrentStreamBufferUsage()
@ -501,6 +509,74 @@ bool NormSenderNode::ReadNextCmd(char* buffer, unsigned int* buflen)
}
} // end NormSenderNode::ReadNextCmd()
bool NormSenderNode::SendAckEx(const char* appAck, unsigned int appAckLen)
{
// First copy in the new appAck content for transmission
if (NULL != appAck)
{
if (appAckLen != ack_ex_length)
{
if (NULL != ack_ex_buffer)
{
delete[] ack_ex_buffer;
ack_ex_buffer = NULL;
ack_ex_length = 0;
}
// Make sure there is room for the header extension
if (appAckLen > SegmentSize())
{
PLOG(PL_ERROR, "NormSenderNode::SendAckEx() error: application-defined ACK_REQ content too large!\n");
ack_ex_pending = false;
return false;
}
else if (NULL == (ack_ex_buffer = new char[appAckLen]))
{
PLOG(PL_ERROR, "NormSenderNode::SendAckEx() new app_req_buffer error: %s\n", GetErrorString());
ack_ex_pending = false;
return false;
}
}
memcpy(ack_ex_buffer, appAck, appAckLen);
ack_ex_length = appAckLen;
}
else if (NULL != ack_ex_buffer)
{
delete[] ack_ex_buffer;
ack_ex_buffer = NULL;
ack_ex_length = 0;
}
ack_ex_pending = false;
if (!ack_timer.IsActive())
OnAckTimeout(ack_timer);
return true;
} // end NormSenderNode::SendAckEx()
bool NormSenderNode::GetWatermarkEx(char* buffer, unsigned int* buflen)
{
if (0 != ack_ex_length)
{
if (NULL != buflen)
{
if (*buflen < ack_ex_length)
{
*buflen = ack_ex_length;
return false;
}
*buflen = ack_ex_length;
if (NULL != buffer)
memcpy(buffer, ack_ex_buffer, ack_ex_length);
else
return false;
}
return true;
}
else
{
if (NULL != buflen) *buflen = 0;
return false; // no application-defined ACK request data
}
} // end NormSenderNode::GetWatermarkEx()
void NormSenderNode::SetRobustFactor(int value)
{
robust_factor = value;
@ -561,9 +637,9 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime,
// build proper NACKs since we have no prior state
fec_id = squelch.GetFecId();
if (2 == fec_id) // see comment in HandleObjectMessage() method on this
fec_m = 16;
fti_data.SetFecFieldSize(16);
else
fec_m = 8;
fti_data.SetFecFieldSize(8);
}
else
{
@ -576,9 +652,7 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime,
NormObject* obj = rx_table.Find(objectId);
if ((NULL != obj) && (NormObject::STREAM == obj->GetType()))
{
NormBlockId blockId = squelch.GetFecBlockId(fec_m);
//NormStreamObject* stream = (NormStreamObject*)obj;
//TRACE("Prune(%u) 8 (read_index = %u.%hu)...\n", (UINT32)blockId, (UINT32)stream->GetNextBlockId(), stream->GetNextSegmentId());
NormBlockId blockId = squelch.GetFecBlockId(fti_data.GetFecFieldSize());
static_cast<NormStreamObject*>(obj)->Prune(blockId, true);
}
// 3) Discard any invalidated objects (those listed in the squelch)
@ -674,7 +748,7 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime,
}
else
{
double nominalSize = nominal_packet_size ? nominal_packet_size : segment_size;
double nominalSize = nominal_packet_size ? nominal_packet_size : SegmentSize();
cc_rate = NormSession::CalculateRate(nominalSize, rtt_estimate, ccLoss);
r = cc_rate / send_rate;
r = MIN(r, 0.9);
@ -724,14 +798,14 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime,
// build proper NACKs since we have no prior state
fec_id = flush.GetFecId();
if (2 == fec_id) // see comment in HandleObjectMessage() method on this
fec_m = 16;
fti_data.SetFecFieldSize(16);
else
fec_m = 8;
fti_data.SetFecFieldSize(8);
}
else if (flush.GetFecId() == fec_id)
{
blockId = flush.GetFecBlockId(fec_m);
symbolId = flush.GetFecSymbolId(fec_m);
blockId = flush.GetFecBlockId(fti_data.GetFecFieldSize());
symbolId = flush.GetFecSymbolId(fti_data.GetFecFieldSize());
}
else
{
@ -759,7 +833,45 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime,
watermark_object_id = objectId;
watermark_block_id = blockId;
watermark_segment_id = symbolId;
if (!ack_timer.IsActive())
// Check for application-extended watermark request (see NormSetWatermarkEx())
const char* appAckReq = NULL;
unsigned int appAckReqLen = 0;
NormAppAckExtension ext;
while (flush.GetNextExtension(ext))
{
if (NormHeaderExtension::APP_ACK == ext.GetType())
{
appAckReq = ext.GetContent();
appAckReqLen = ext.GetContentLength();
}
}
if (NULL != appAckReq)
{
// We need to bubble this up to the application before we acknowledge watermark
// so app can set any extended ACK content in response
if (appAckReqLen != ack_ex_length)
{
if (NULL != ack_ex_buffer) delete[] ack_ex_buffer;
if (NULL == (ack_ex_buffer = new char[appAckReqLen]))
{
// TBD - notify app of allocation error
PLOG(PL_ERROR, "NormSenderNode::HandleCommand() new ack_ex_buffer error: %s\n", GetErrorString());
ack_ex_length = 0;
}
else
{
ack_ex_length = appAckReqLen;
}
}
if (NULL != ack_ex_buffer)
{
memcpy(ack_ex_buffer, appAckReq, appAckReqLen);
ack_ex_pending = true;
session.Notify(NormController::RX_ACK_REQUEST, this, NULL);
}
}
else if (!ack_timer.IsActive())
{
double ackBackoff = (session.Address().IsMulticast() && (backoff_factor > 0.0)) ?
UniformRand(grtt_estimate) : 0.0;
@ -820,8 +932,8 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime,
else
{
unsigned int cmdLength = appCmd.GetContentLength();
if ((cmdLength <= segment_size) ||
((0 == segment_size) && (cmdLength < 8192)))
if ((cmdLength <= SegmentSize()) ||
((0 == SegmentSize()) && (cmdLength < 8192)))
{
// 2) Copy the app-defined command content into our buffer
buf->SetContent(appCmd.GetContent(), appCmd.GetContentLength());
@ -863,7 +975,7 @@ void NormSenderNode::HandleCCFeedback(UINT8 ccFlags, double ccRate)
{
// We're suppressed by non-CLR receivers with no RTT confirmed
// and/or lower rate
double nominalSize = nominal_packet_size ? nominal_packet_size : segment_size;
double nominalSize = nominal_packet_size ? nominal_packet_size : SegmentSize();
double ccLoss = slow_start ? 0.0 : LossEstimate();
double localRate = (0.0 == ccLoss) ?
(2.0*recv_rate) :
@ -988,7 +1100,7 @@ void NormSenderNode::HandleRepairContent(const UINT32* buffer, UINT16 bufferLen)
}
bool repairInfo = req.FlagIsSet(NormRepairRequest::INFO);
NormRepairRequest::Iterator iterator(req, fec_id, fec_m); // assumes constant "m"
NormRepairRequest::Iterator iterator(req, fec_id, fti_data.GetFecFieldSize()); // assumes constant "m"
NormObjectId nextObjectId, lastObjectId;
NormBlockId nextBlockId, lastBlockId;
UINT16 nextBlockLen, lastBlockLen;
@ -1119,8 +1231,6 @@ void NormSenderNode::DeleteObject(NormObject* obj)
NormBlock* NormSenderNode::GetFreeBlock(NormObjectId objectId, NormBlockId blockId)
{
NormBlock* b = block_pool.Get();
//TRACE("enter NormSenderNode::GetFreeBlock() empty:%d\n", (NULL == b));
if (NULL == b)
{
if (session.ReceiverIsSilent() || session.RcvrIsRealtime())
@ -1259,7 +1369,7 @@ bool NormSenderNode::PreallocateRxStream(unsigned int bufferSize,
return true;
} // end NormSenderNode::PreallocateRxStream()
// TBD - Move this method to NormObjectMsg class
bool NormSenderNode::GetFtiData(const NormObjectMsg& msg, NormFtiData& ftiData)
{
UINT8 fecId = msg.GetFecId();
@ -1358,31 +1468,36 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
bool allocateBuffers = true;
bool gotFTI = false;
NormFtiData ftiData;
if (BuffersAllocated())
{
// Validate that allocated buffers match object FEC params
if (fecId == fec_id)
{
if (GetFtiData(msg, ftiData))
if (GetFtiData(msg, ftiData) || session.GetPresetFtiData(ftiData))
{
gotFTI = true;
if ((ftiData.GetSegmentSize() != segment_size) ||
(ftiData.GetFecFieldSize() != fec_m) ||
(ftiData.GetFecMaxBlockLen() != ndata) ||
(ftiData.GetFecNumParity() != nparity))
if ((ftiData.GetSegmentSize() != SegmentSize()) ||
(ftiData.GetFecFieldSize() != fti_data.GetFecFieldSize()) ||
(ftiData.GetFecMaxBlockLen() != fti_data.GetFecMaxBlockLen()) ||
(ftiData.GetFecNumParity() != fti_data.GetFecNumParity()))
{
FreeBuffers(); // force reallocation because fec params changed
fti_data = ftiData;
}
else
{
allocateBuffers = false; // FEC params match
}
}
else if ((NormMsg::INFO != msgType) && msg.FlagIsSet(NormObjectMsg::FLAG_INFO))
{
// This handles case where only NORM_INFO carries FTI Info to reduce overhead
// We have to assume sender FTI hasn't changed ...
allocateBuffers = false;
}
else
{
// TBD - handle case where only NORM_INFO carries FTI Info to reduce overhead
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu - no FTI provided!\n",
(unsigned long)LocalNodeId(), (unsigned long)GetId());
return; // (TBD) notify app of error ??
@ -1392,29 +1507,37 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
{
FreeBuffers(); // force reallocation because fec id changed
}
}
} // end if (BuffersAllocated())
NormBlockId blockId;
NormSegmentId segmentId;
if (NormMsg::INFO == msgType)
{
fec_id = fecId;
// This assumes fec_id 2 (Reed Solomon) is only used when
// 16-bit RS code is needed (otherwise fec_id 5 should have been used)
// This is a limitation of the current RS specification where fec_m is
// only provided in FEC OTI. (Note we could dig into FEC OTI of NORM_INFO
// message, but FLUSH/SQUELCH commands can't do this anyway and because we
// defer buffer allocation until NORM_DATA is received, it doesn't really matter
// (Any NACKs generate will have zero-value FEC blockId / symbolId anyway)
if (2 == fecId)
fec_m = 16;
if (!BuffersAllocated())
{
fec_id = fecId;
// Go ahead and capture FTI from INFO
if (GetFtiData(msg, ftiData) || session.GetPresetFtiData(ftiData))
{
gotFTI = true;
fti_data = ftiData;
}
else
{
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu - no FTI provided!\n",
(unsigned long)LocalNodeId(), (unsigned long)GetId());
return; // (TBD) notify app of error ??
}
}
else
fec_m = 8;
{
ASSERT(gotFTI);
}
blockId = 0;
segmentId = 0;
}
else
else // NormMsg::DATA
{
if (allocateBuffers)
{
@ -1430,16 +1553,26 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
{
gotFTI = true;
}
else
else if (fti_data.IsValid())
{
ftiData = fti_data;
gotFTI = true;
}
else if (session.GetPresetFtiData(ftiData))
{
gotFTI = true;
}
else if ((NormMsg::INFO != msgType) && !msg.FlagIsSet(NormObjectMsg::FLAG_INFO))
{
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu - no FTI provided!\n",
(unsigned long)LocalNodeId(), (unsigned long)GetId());
// (TBD) notify app of error ??
return;
}
// else wait for NORM_INFO message with sender FTI
}
if (!AllocateBuffers(session.RemoteSenderBufferSize(), fecId, ftiData.GetFecInstanceId(), ftiData.GetFecFieldSize(),
ftiData.GetSegmentSize(), ftiData.GetFecMaxBlockLen(), ftiData.GetFecNumParity()))
if (gotFTI && !AllocateBuffers(session.RemoteSenderBufferSize(), fecId, ftiData.GetFecInstanceId(), ftiData.GetFecFieldSize(),
ftiData.GetSegmentSize(), ftiData.GetFecMaxBlockLen(), ftiData.GetFecNumParity()))
{
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu buffer allocation error\n",
(unsigned long)LocalNodeId(), (unsigned long)GetId());
@ -1448,10 +1581,23 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
}
} // end if (allocateBuffers)
ASSERT(0 != fec_m);
const NormDataMsg& data = static_cast<const NormDataMsg&>(msg);
blockId = data.GetFecBlockId(fec_m);
segmentId = data.GetFecSymbolId(fec_m);
if (fti_data.IsValid())
{
ASSERT(0 != fti_data.GetFecFieldSize());
const NormDataMsg& data = static_cast<const NormDataMsg&>(msg);
blockId = data.GetFecBlockId(fti_data.GetFecFieldSize());
segmentId = data.GetFecSymbolId(fti_data.GetFecFieldSize());
}
else
{
// These won't come into play anyway
if (2 == fecId)
fti_data.SetFecFieldSize(16);
else
fti_data.SetFecFieldSize(8);
blockId = 0;
segmentId = 0;
}
} // end if/else (NormMsg::INFO == msgType)
ObjectStatus status;
@ -1485,16 +1631,41 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
}
bool presetStream = false;
NormObject* obj = NULL;
bool doInsert = true;
switch (status)
{
case OBJ_PENDING:
if (NULL != (obj = rx_table.Find(objectId)))
break;
{
if (NULL != (obj = rx_table.Find(objectId)))
{
if (0 == obj->GetSize().GetOffset())
{
// It's a seen object for which are awaiting FTI
if (GetFtiData(msg, ftiData))
{
gotFTI = true;
obj->SetNackingMode(default_nacking_mode);
doInsert = false;
// Intentionally pass through to case OBJ_NEW
}
else
{
obj = NULL; // keep waiting for FTI
break;
}
}
else
{
break; // handle as normal pending object
}
}
// else intentionally pass through to case OBJ_NEW
}
case OBJ_NEW:
{
if (msg.FlagIsSet(NormObjectMsg::FLAG_STREAM))
{
if (NULL != preset_stream)
if ((NULL != preset_stream) && ((NULL == obj) || (obj == static_cast<NormObject*>(preset_stream))))
{
obj = static_cast<NormObject*>(preset_stream);
// Validate FTI params
@ -1505,7 +1676,11 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
{
gotFTI = true;
}
else
else if (session.GetPresetFtiData(ftiData))
{
gotFTI = true;
}
else if ((NormMsg::INFO != msgType) && !msg.FlagIsSet(NormObjectMsg::FLAG_INFO))
{
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu - no FTI provided!\n",
(unsigned long)LocalNodeId(), (unsigned long)GetId());
@ -1513,12 +1688,13 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
return;
}
}
if ((obj->GetSize() != ftiData.GetObjectSize()) ||
(obj->GetFecId() != fecId) ||
(obj->GetSegmentSize() != ftiData.GetSegmentSize()) ||
(obj->GetFecMaxBlockLen() != ftiData.GetFecMaxBlockLen()) ||
(obj->GetFecNumParity() != ftiData.GetFecNumParity()) ||
(obj->GetFecFieldSize() != ftiData.GetFecFieldSize()))
if (gotFTI &&
((obj->GetSize() != ftiData.GetObjectSize()) ||
(obj->GetFecId() != fecId) ||
(obj->GetSegmentSize() != ftiData.GetSegmentSize()) ||
(obj->GetFecMaxBlockLen() != ftiData.GetFecMaxBlockLen()) ||
(obj->GetFecNumParity() != ftiData.GetFecNumParity()) ||
(obj->GetFecFieldSize() != ftiData.GetFecFieldSize())))
{
PLOG(PL_WARN, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu warning: "
"FTI does not match preset_stream!\n",
@ -1543,7 +1719,7 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
}
}
}
else if (msg.FlagIsSet(NormObjectMsg::FLAG_FILE))
else if (msg.FlagIsSet(NormObjectMsg::FLAG_FILE) && (NULL == obj))
{
#ifdef SIMULATE
if (!(obj = new NormSimObject(session, this, objectId)))
@ -1552,10 +1728,10 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
#endif
{
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() new NORM_OBJECT_FILE error: %s\n",
GetErrorString());
GetErrorString());
}
}
else
else if (NULL == obj)
{
if (!(obj = new NormDataObject(session, this, objectId, session.GetSessionMgr().GetDataFreeFunction())))
{
@ -1571,17 +1747,14 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
{
ASSERT(rx_table.CanInsert(objectId));
ASSERT(rx_pending_mask.Test(objectId));
rx_table.Insert(obj);
if (doInsert) rx_table.Insert(obj);
// Pull out FTI parameters from header extension if we didn't get it above
if (!gotFTI && !GetFtiData(msg, ftiData))
if (!gotFTI)
{
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu "
"new obj>%hu - no FTI provided!\n", (unsigned long)LocalNodeId(),
(unsigned long)GetId(), (UINT16)objectId);
if (!presetStream) DeleteObject(obj);
obj = NULL;
if (GetFtiData(msg, ftiData) || session.GetPresetFtiData(ftiData))
gotFTI = true;
}
else
if (gotFTI)
{
if (presetStream ||
obj->RxOpen(ftiData.GetObjectSize(),
@ -1641,6 +1814,23 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
obj = NULL;
}
}
else if ((NormMsg::INFO != msgType) && msg.FlagIsSet(NormObjectMsg::FLAG_INFO))
{
// Open a zero-sized object in NACK_INFO_ONLY nacking mode until NORM_INFO w/ FTI arrives
obj->SetPendingInfo(true, fecId);
obj->SetNackingMode(NormObject::NACK_INFO_ONLY);
if (presetStream) preset_stream = NULL; // we're using it up
obj = NULL; // can't process NORM_DATA until we have FTI
// TBD - buffer received messages instead of discarding them???
}
else
{
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu "
"new obj>%hu - no FTI provided!\n", (unsigned long)LocalNodeId(),
(unsigned long)GetId(), (UINT16)objectId);
if (!presetStream) DeleteObject(obj);
obj = NULL;
}
}
break;
}
@ -1692,7 +1882,7 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
// Optional "object boundary repair check (non-streams only!)
// (checks for repair needs for objects _prior_ to current objectId)
// (also requests "info" for current objectId)
if (obj && obj->IsStream())
if (NULL != obj && obj->IsStream())
RepairCheck(NormObject::TO_BLOCK, objectId, blockId, segmentId);
else
RepairCheck(NormObject::THRU_INFO, objectId, blockId, segmentId);
@ -1711,7 +1901,7 @@ bool NormSenderNode::SyncTest(const NormObjectMsg& msg) const
bool result = msg.FlagIsSet(NormObjectMsg::FLAG_STREAM);
// Allow sync on INFO or block zero DATA message
result = result || (NormMsg::INFO == msg.GetType()) ?
true : (NormBlockId(0) == ((const NormDataMsg&)msg).GetFecBlockId(fec_m));
true : (NormBlockId(0) == ((const NormDataMsg&)msg).GetFecBlockId(fti_data.GetFecFieldSize()));
// Never sync on repair messages
result = result && !msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR);
return result;
@ -2142,7 +2332,7 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
return false;
}
nack->Init();
UINT16 payloadMax = 4*segment_size;
UINT16 payloadMax = 4*SegmentSize();
// If we sync'd to non-DATA, we don't yet know the sender segment_size
if (0 == payloadMax)
payloadMax = 4*NormNackMsg::DEFAULT_LENGTH_MAX;
@ -2176,7 +2366,7 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
}
else
{
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double nominalSize = (nominal_packet_size > SegmentSize()) ? nominal_packet_size : SegmentSize();
if (0 == nominalSize) nominalSize = 512; // TBD - what should this really be
double ccRate = NormSession::CalculateRate(nominalSize,
rtt_estimate,
@ -2273,13 +2463,13 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
switch (nextForm)
{
case NormRepairRequest::ITEMS:
req.AppendRepairItem(fec_id, fec_m, prevId, 0, ndata, 0);
req.AppendRepairItem(fec_id, fti_data.GetFecFieldSize(), prevId, 0, BlockSize(), 0);
if (2 == consecutiveCount)
req.AppendRepairItem(fec_id, fec_m, prevId+1, 0, ndata, 0);
req.AppendRepairItem(fec_id, fti_data.GetFecFieldSize(), prevId+1, 0, BlockSize(), 0);
break;
case NormRepairRequest::RANGES:
req.AppendRepairRange(fec_id, fec_m, prevId, 0, ndata, 0,
prevId+consecutiveCount-1, 0, ndata, 0);
req.AppendRepairRange(fec_id, fti_data.GetFecFieldSize(), prevId, 0, BlockSize(), 0,
prevId+consecutiveCount-1, 0, BlockSize(), 0);
break;
default:
break;
@ -2340,7 +2530,7 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
ASSERT(nack->GetRepairContentLength() > 0);
if (!session.ReceiverIsSilent())
{
UINT16 singleNackSize = segment_size ? segment_size : NormNackMsg::DEFAULT_LENGTH_MAX;
UINT16 singleNackSize = SegmentSize() ? SegmentSize() : NormNackMsg::DEFAULT_LENGTH_MAX;
if (nack->GetRepairContentLength() <= singleNackSize)
{
session.SendMessage(*nack);
@ -2394,7 +2584,7 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
}
// Uncommenting the line below treats ((0 == nparity) && 0.0 == backoff_factor)
// as a special case (assumes zero sender aggregateInterval)
holdoffInterval = ((0 != nparity) || (backoff_factor > 0.0)) ? holdoffInterval : grtt_estimate;
holdoffInterval = ((0 != NumParity()) || (backoff_factor > 0.0)) ? holdoffInterval : grtt_estimate;
repair_timer.SetInterval(holdoffInterval);
PLOG(PL_DEBUG, "NormSenderNode::OnRepairTimeout() node>%lu sender>%lu begin NACK hold-off: %lf sec ...\n",
(unsigned long)LocalNodeId(), (unsigned long)GetId(), holdoffInterval);
@ -2442,23 +2632,23 @@ void NormSenderNode::FragmentNack(NormNackMsg& superNack)
{
const UINT16 REQ_HDR_LEN = 4; // TBD - get from normMessage.h instead
requestOffset += requestLength;
if ((payloadLength + requestLength) <= segment_size)
if ((payloadLength + requestLength) <= SegmentSize())
{
// Copy whole request over
nack->AppendRepairRequest(superReq);
payloadLength += requestLength;
}
else if ((payloadLength + REQ_HDR_LEN) < segment_size)
else if ((payloadLength + REQ_HDR_LEN) < SegmentSize())
{
// Duplicate request and add individual repair items
NormRepairRequest::Form requestForm = superReq.GetForm();
NormRepairRequest req;
nack->AttachRepairRequest(req, segment_size);
nack->AttachRepairRequest(req, SegmentSize());
req.SetForm(requestForm);
req.SetFlags(superReq.GetFlags());
payloadLength += REQ_HDR_LEN;
NormRepairRequest::Iterator iterator(superReq, fec_id, fec_m);
NormRepairRequest::Iterator iterator(superReq, fec_id, fti_data.GetFecFieldSize());
NormObjectId objectId, lastObjectId;
NormBlockId blockId, lastBlockId;
UINT16 blockLen, lastBlockLen;
@ -2471,24 +2661,24 @@ void NormSenderNode::FragmentNack(NormNackMsg& superNack)
itemLength += iterator.NextRepairItem(&lastObjectId, &lastBlockId,
&lastBlockLen, &lastSegmentId);
}
if ((payloadLength + itemLength) > segment_size)
if ((payloadLength + itemLength) > SegmentSize())
{
// We have filled the NACK, so pack, send, and reset request
nack->PackRepairRequest(req);
session.SendMessage(*nack);
nack_count++;
nack->ResetPayload();
nack->AttachRepairRequest(req, segment_size);
nack->AttachRepairRequest(req, SegmentSize());
payloadLength = REQ_HDR_LEN;
}
if (NormRepairRequest::RANGES == requestForm)
{
req.AppendRepairRange(fec_id, fec_m, objectId, blockId, blockLen, segmentId,
req.AppendRepairRange(fec_id, fti_data.GetFecFieldSize(), objectId, blockId, blockLen, segmentId,
lastObjectId, lastBlockId, lastBlockLen, lastSegmentId);
}
else
{
req.AppendRepairItem(fec_id, fec_m, objectId, blockId, blockLen, segmentId);
req.AppendRepairItem(fec_id, fti_data.GetFecFieldSize(), objectId, blockId, blockLen, segmentId);
}
payloadLength += itemLength;
}
@ -2534,7 +2724,7 @@ void NormSenderNode::UpdateRecvRate(const struct timeval& currentTime, unsigned
if ((interval >= measurementInterval) && (currentRecvRate < recv_rate))
{
// Make sure we've allowed sufficient time for a measurement at low rates
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double nominalSize = (nominal_packet_size > SegmentSize()) ? nominal_packet_size : SegmentSize();
double minInterval = 4.0 * nominalSize / recv_rate;
if (measurementInterval < minInterval) measurementInterval = minInterval;
}
@ -2700,7 +2890,7 @@ bool NormSenderNode::UpdateLossEstimate(const struct timeval& currentTime,
{
// Calculate loss initialization based on current receive rate
// and rtt estimation
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double nominalSize = (nominal_packet_size > SegmentSize()) ? nominal_packet_size : SegmentSize();
double lossInit = nominalSize / (recv_rate * rtt_estimate);
lossInit *= lossInit;
lossInit *= 3.0/2.0;
@ -2767,7 +2957,7 @@ void NormSenderNode::AttachCCFeedback(NormAckMsg& ack)
else
{
//double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double nominalSize = (0 != nominal_packet_size) ? nominal_packet_size : segment_size;
double nominalSize = (0 != nominal_packet_size) ? nominal_packet_size : SegmentSize();
double ccRate = NormSession::CalculateRate(nominalSize, rtt_estimate, ccLoss);
#ifdef LIMIT_CC_RATE
// Experimental modification to NORM-CC where congestion control rate is limited
@ -2795,7 +2985,7 @@ void NormSenderNode::AttachCCFeedback(NormAckMsg& ack)
bool NormSenderNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
{
// Build and send NORM_ACK(CC)
if (ack_pending && (1 == cc_timer.GetRepeatCount()))
if (ack_pending && !ack_ex_pending && (1 == cc_timer.GetRepeatCount()))
{
// Send ACK flush right away (CC feedback is included)
if (ack_timer.IsActive()) ack_timer.Deactivate();
@ -2869,6 +3059,8 @@ bool NormSenderNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
bool NormSenderNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
{
// Build and send NORM_ACK(FLUSH)
if (ack_ex_pending)
return true; // Will acknowledge when application services RX_ACK_REQUEST notification
NormAckFlushMsg* ack = (NormAckFlushMsg*)session.GetMessageFromPool();
if (NULL != ack)
{
@ -2878,6 +3070,14 @@ bool NormSenderNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
ack->SetAckType(NormAck::FLUSH);
ack->SetAckId(0);
AttachCCFeedback(*ack);
if (0 != ack_ex_length)
{
NormAppAckExtension ext;
ack->AttachExtension(ext);
ext.SetContent(ack_ex_buffer, ack_ex_length);
ack->PackExtension(ext);
}
ack->SetObjectId(watermark_object_id);
// _Attempt_ to set the fec_payload_id source block length field appropriately
@ -2885,11 +3085,12 @@ bool NormSenderNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
NormObject* obj = rx_table.Find(watermark_object_id);
if (NULL != obj)
blockLen = obj->GetBlockSize(watermark_block_id);
else if (watermark_segment_id < ndata)
blockLen = ndata;
else if (watermark_segment_id < BlockSize())
blockLen = BlockSize();
else
blockLen = watermark_segment_id;
ack->SetFecPayloadId(fec_id, watermark_block_id.GetValue(), watermark_segment_id, blockLen, fec_m);
ack->SetFecPayloadId(fec_id, watermark_block_id.GetValue(), watermark_segment_id, blockLen, fti_data.GetFecFieldSize());
if (unicast_nacks)
ack->SetDestination(GetAddress());
@ -2932,15 +3133,67 @@ bool NormSenderNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
NormAckingNode::NormAckingNode(class NormSession& theSession, NormNodeId nodeId)
: NormNode(ACKER, theSession, nodeId), ack_received(false), req_count(theSession.GetTxRobustFactor())
{
: NormNode(ACKER, theSession, nodeId),
ack_received(false), req_count(theSession.GetTxRobustFactor()),
ack_ex_buffer(NULL), ack_ex_length(0)
{
}
NormAckingNode::~NormAckingNode()
{
if (NULL != ack_ex_buffer)
{
delete[] ack_ex_buffer;
ack_ex_buffer = NULL;
ack_ex_length = 0;
}
}
bool NormAckingNode::SetAckEx(const char* buffer, UINT16 numBytes)
{
if (numBytes != ack_ex_length)
{
if (NULL != ack_ex_buffer) delete[] ack_ex_buffer;
if (NULL == (ack_ex_buffer = new char[numBytes]))
{
// TBD - notify app of errror
PLOG(PL_ERROR, "NormAckingNode::SetAppAckContent() new ack_ex_buffer error: %s\n", GetErrorString());
ack_ex_length = 0;
return false;
}
ack_ex_length = numBytes;
}
memcpy(ack_ex_buffer, buffer, numBytes);
return true;
} // end NormAckingNode::SetAckEx()
bool NormAckingNode::GetAckEx(char* buffer, unsigned int* buflen)
{
if (0 != ack_ex_length)
{
if (NULL != buflen)
{
if (*buflen < ack_ex_length)
{
*buflen = ack_ex_length;
return false;
}
*buflen = ack_ex_length;
if (NULL != buffer)
memcpy(buffer, ack_ex_buffer, ack_ex_length);
else
return false;
}
return true;
}
else
{
if (NULL != buflen) *buflen = 0;
return false; // no application-defined ACK request data
}
} // end NormAckingNode::GetAckEx()
NormNodeTree::NormNodeTree()
: root(NULL)
{
@ -3449,7 +3702,6 @@ bool NormLossEstimator2::Update(const struct timeval& currentTime,
seeking_loss_event = CONFIRMING; // wait for more loss to confirm congestion event
event_time = event_time_orig = currentTime;
outageDepth = 0;
//TRACE("Deferring loss event for confirmation ...\n");
}
}
@ -3466,7 +3718,6 @@ bool NormLossEstimator2::Update(const struct timeval& currentTime,
discount[i] = discount[i-1]*current_discount;
discount[0] = 1.0;
current_discount = 1.0;
//TRACE("Confirmed loss event ...(tolerate:%d)\n", tolerate_loss);
seeking_loss_event = CONFIRMED;
event_time = event_time_orig = currentTime;
newLossEvent = true;

140
src/common/normObject.cpp
View File

@ -50,8 +50,6 @@ void NormObject::Retain()
void NormObject::Release()
{
//if ((NULL != sender) && (reference_count == 1))
// TRACE("NormObject final release node>%lu object>%hu\n", (unsigned long)sender->GetId(), (unsigned short)transport_id);
if (sender) sender->Release();
if (reference_count)
{
@ -93,18 +91,18 @@ bool NormObject::Open(const NormObjectSize& objectSize,
if (infoLen > 0)
{
pending_info = true;
info_len = 0;
if (!(info_ptr = new char[segmentSize]))
{
PLOG(PL_FATAL, "NormObject::Open() info allocation error\n");
return false;
}
}
info_len = 0; // will be set properly upon NORM_INFO arrival
last_nack_time.GetCurrentTime(); // init to now
}
else
{
if (infoPtr)
if (NULL != infoPtr)
{
if (NULL != info_ptr) delete[] info_ptr;
if (infoLen > segmentSize)
@ -123,6 +121,14 @@ bool NormObject::Open(const NormObjectSize& objectSize,
info_len = infoLen;
pending_info = true;
}
else
{
if (NormSession::FTI_INFO == session.SenderFtiMode())
{
pending_info = true;
}
}
}
// Compute number of segments and coding blocks for the object
@ -190,11 +196,11 @@ bool NormObject::Open(const NormObjectSize& objectSize,
}
else
{
// Set everything pending
// Set everything pending (i.e, nothing sent or received yet)
pending_mask.Clear();
pending_mask.SetBits(0, numBlocks.LSB());
// Compute FEC block structure per NORM Protocol Spec Section 5.1.1
// (Note NormObjectSize divide operator always rounds _up_)
// (Note NormObjectSize divide operator always rounds _up_, i.e., ceil(numSegments/numBlocks))
NormObjectSize largeBlockSize = numSegments / numBlocks;
ASSERT(0 == largeBlockSize.MSB());
large_block_size = largeBlockSize.LSB();
@ -290,7 +296,7 @@ NormObjectSize NormObject::GetBytesPending() const
bool NormObject::HandleInfoRequest(bool holdoff)
{
bool increasedRepair = false;
if (info_ptr)
if (info_ptr || (NormSession::FTI_INFO == session.SenderFtiMode()))
{
if (!repair_info)
{
@ -379,7 +385,7 @@ bool NormObject::HandleBlockRequest(const NormBlockId& firstId, const NormBlockI
bool NormObject::TxReset(NormBlockId firstBlock, bool requeue)
{
bool increasedRepair = false;
if (!pending_info && HaveInfo())
if (!pending_info && (HaveInfo() || (NormSession::FTI_INFO == session.SenderFtiMode())))
{
increasedRepair = true;
pending_info = true;
@ -1387,9 +1393,22 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
"Warning! info too long.\n", (unsigned long)LocalNodeId(),
(unsigned long)sender->GetId(), (UINT16)transport_id);
}
memcpy(info_ptr, infoMsg.GetInfo(), info_len);
pending_info = false;
session.Notify(NormController::RX_OBJECT_INFO, sender, this);
else if (0 == info_len)
{
// NORM_INFO used to convey FTI only
if (NULL != info_ptr)
{
delete[] info_ptr;
info_ptr = NULL;
}
pending_info = false;
}
else
{
memcpy(info_ptr, infoMsg.GetInfo(), info_len);
pending_info = false;
session.Notify(NormController::RX_OBJECT_INFO, sender, this);
}
}
else
{
@ -1538,7 +1557,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
if (nextErasure < numData)
{
// Use "NormObject::RetrieveSegment() method to "retrieve"
// source symbol segments already received which weren't cached.
// source symbol segments already received which aren't still cached.
for (UINT16 nextSegment = 0; nextSegment < numData; nextSegment++)
{
if (block->IsPending(nextSegment))
@ -1622,7 +1641,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
// so it's not called unnecessarily
if (objectUpdated && notify_on_update)
{
if ((NULL == stream) || stream->DetermineReadReadiness())
if ((NULL == stream) || stream->DetermineReadReadiness() || session.RcvrIsLowDelay())
{
notify_on_update = false;
session.Notify(NormController::RX_OBJECT_UPDATED, sender, this);
@ -1780,50 +1799,55 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
default:
break;
}
if (info_ptr)
NormSession::FtiMode ftiMode = session.SenderFtiMode();
if ((NULL != info_ptr) || (NormSession::FTI_INFO == ftiMode))
msg->SetFlag(NormObjectMsg::FLAG_INFO);
msg->SetObjectId(transport_id);
// We currently always apply the FTI extension
switch (fec_id)
// Apply FTI extension as needed
if ((NormSession::FTI_ALWAYS == ftiMode) ||
(pending_info && (NormSession::FTI_INFO == ftiMode)))
{
case 2:
switch (fec_id)
{
NormFtiExtension2 fti;
msg->AttachExtension(fti);
fti.SetObjectSize(object_size);
fti.SetFecFieldSize(fec_m);
fti.SetFecGroupSize(1);
fti.SetSegmentSize(segment_size);
fti.SetFecMaxBlockLen(ndata);
fti.SetFecNumParity(nparity);
break;
case 2:
{
NormFtiExtension2 fti;
msg->AttachExtension(fti);
fti.SetObjectSize(object_size);
fti.SetFecFieldSize(fec_m);
fti.SetFecGroupSize(1);
fti.SetSegmentSize(segment_size);
fti.SetFecMaxBlockLen(ndata);
fti.SetFecNumParity(nparity);
break;
}
case 5:
{
NormFtiExtension5 fti;
msg->AttachExtension(fti);
fti.SetObjectSize(object_size);
fti.SetSegmentSize(segment_size);
fti.SetFecMaxBlockLen((UINT8)ndata);
fti.SetFecNumParity((UINT8)nparity);
break;
}
case 129:
{
NormFtiExtension129 fti;
msg->AttachExtension(fti);
fti.SetObjectSize(object_size);
fti.SetFecInstanceId(0); // ZERO is for legacy MDP/NORM FEC encoder (TBD - use appropriate instanceId)
fti.SetSegmentSize(segment_size);
fti.SetFecMaxBlockLen(ndata);
fti.SetFecNumParity(nparity);
break;
}
default:
ASSERT(0);
return false;
}
case 5:
{
NormFtiExtension5 fti;
msg->AttachExtension(fti);
fti.SetObjectSize(object_size);
fti.SetSegmentSize(segment_size);
fti.SetFecMaxBlockLen((UINT8)ndata);
fti.SetFecNumParity((UINT8)nparity);
break;
}
case 129:
{
NormFtiExtension129 fti;
msg->AttachExtension(fti);
fti.SetObjectSize(object_size);
fti.SetFecInstanceId(0); // ZERO is for legacy MDP/NORM FEC encoder (TBD - use appropriate instanceId)
fti.SetSegmentSize(segment_size);
fti.SetFecMaxBlockLen(ndata);
fti.SetFecNumParity(nparity);
break;
}
default:
ASSERT(0);
return false;
}
if (pending_info)
{
@ -1990,7 +2014,7 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
block->ResetParityCount(nparity);
pending_mask.Unset(blockId.GetValue());
// for EMCON sending, mark NORM_INFO for re-transmission, if applicable
if (session.SndrEmcon() && HaveInfo())
if (session.SndrEmcon() && (HaveInfo() || (NormSession::FTI_INFO == session.SenderFtiMode())))
pending_info = true;
}
block = NULL;
@ -2052,7 +2076,7 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
block->ResetParityCount(nparity);
pending_mask.Unset(blockId.GetValue());
// for EMCON sending, mark NORM_INFO for re-transmission, if applicable
if (session.SndrEmcon() && HaveInfo())
if (session.SndrEmcon() && (HaveInfo() || (NormSession::FTI_INFO == session.SenderFtiMode())))
pending_info = true;
// Advance sender use of "max_pending_block" so we always
// know when a block should be flagged as IN_REPAIR
@ -2506,7 +2530,6 @@ NormDataObject::NormDataObject(class NormSession& theSession,
NormDataObject::~NormDataObject()
{
//TRACE("NormDataObject destructor ...\n");
Close();
if (data_released)
{
@ -3289,7 +3312,6 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
block->EmptyToPool(segment_pool);
block_pool.Put(block);
block = NULL;
//TRACE("Prune(%u) 6 ...\n", (unsigned int)read_index.block.GetValue());
Prune(read_index.block, false);
break;
}
@ -3315,7 +3337,6 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
Increment(read_index.block);
read_index.segment = 0;
read_index.offset = 0;
//TRACE("Prune(%lu) 7 ...\n", (unsigned long)read_index.block.GetValue());
Prune(read_index.block, false);
}
}
@ -3553,7 +3574,9 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
{
INT32 delta = (INT32)Difference(max_pending_block, read_index.block);
if (delta > session.RcvrGetMaxDelay())
forceForward = true;
{
forceForward = true;
}
}
}
else
@ -3571,7 +3594,6 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
Increment(read_index.block);
read_index.segment = 0;
read_index.offset = 0;
//TRACE("Prune(%u) 1 ...\n", (unsigned int)read_index.block.GetValue());
if (!seekMsgStart) brokenStream = true;
Prune(read_index.block, false);
continue;
@ -3638,7 +3660,6 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
Increment(read_index.block);
read_index.segment = 0;
read_index.offset = 0;
//TRACE("Prune(%u) 2 ...\n", (unsigned int)read_index.block.GetValue());
Prune(read_index.block, false); // prevents repair requests for data we
} // no longer care about (i.e, prior to this block)
if (!seekMsgStart) brokenStream = true;
@ -3712,7 +3733,6 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
Increment(read_index.block);
read_index.segment = 0;
read_index.offset = 0;
//TRACE("Prune(%u) 3 ...\n", (unsigned int)read_index.block.GetValue());
Prune(read_index.block, false);
}
*buflen = 0;
@ -3800,7 +3820,6 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
block_pool.Put(block);
Increment(read_index.block);
read_index.segment = 0;
//TRACE("Prune(%u) 4 ...\n", (unsigned int)read_index.block.GetValue());
Prune(read_index.block, false);
}
continue;
@ -3843,7 +3862,6 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
block_pool.Put(block);
Increment(read_index.block);
read_index.segment = 0;
//TRACE("Prune(%u) 5 ...\n", (unsigned int)read_index.block.GetValue());
Prune(read_index.block, false);
if (0 == bytesToRead)
read_ready = DetermineReadReadiness();

149
src/common/normSession.cpp
View File

@ -48,7 +48,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
backoff_factor(DEFAULT_BACKOFF_FACTOR), is_sender(false),
tx_robust_factor(DEFAULT_ROBUST_FACTOR), instance_id(0),
ndata(DEFAULT_NDATA), nparity(DEFAULT_NPARITY), auto_parity(0), extra_parity(0),
sndr_emcon(false), tx_only(false), tx_connect(false), encoder(NULL),
sndr_emcon(false), tx_only(false), tx_connect(false), fti_mode(FTI_ALWAYS), encoder(NULL),
next_tx_object_id(0),
tx_cache_count_min(DEFAULT_TX_CACHE_MIN),
tx_cache_count_max(DEFAULT_TX_CACHE_MAX),
@ -67,6 +67,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
cc_enable(false), cc_adjust(true), cc_sequence(0), cc_slow_start(true), cc_active(false),
flow_control_factor(DEFAULT_FLOW_CONTROL_FACTOR),
cmd_count(0), cmd_buffer(NULL), cmd_length(0), syn_status(false),
ack_ex_buffer(NULL), ack_ex_length(0),
is_receiver(false), rx_robust_factor(DEFAULT_ROBUST_FACTOR), preset_sender(NULL), unicast_nacks(false),
receiver_silent(false), rcvr_ignore_info(false), rcvr_max_delay(-1), rcvr_realtime(false),
default_repair_boundary(NormSenderNode::BLOCK_BOUNDARY),
@ -721,6 +722,22 @@ bool NormSession::StartSender(UINT16 instanceId,
UINT16 numParity,
UINT8 fecId)
{
UINT16 blockSize = numData + numParity;
if (blockSize <= 255)
fec_m = 8;
else
fec_m = 16;
if (preset_fti.IsValid())
{
if ((preset_fti.GetSegmentSize() != segmentSize) ||
(preset_fti.GetFecMaxBlockLen() != numData) ||
(preset_fti.GetFecNumParity() != numParity) ||
(preset_fti.GetFecFieldSize() != fec_m))
{
PLOG(PL_FATAL, "NormSession::StartSender() preset FTI mismatch error!\n");
return false;
}
}
if (!IsOpen())
{
if (!Open()) return false;
@ -745,7 +762,6 @@ bool NormSession::StartSender(UINT16 instanceId,
}
// Calculate how much memory each buffered block will require
UINT16 blockSize = numData + numParity;
unsigned long maskSize = blockSize >> 3;
if (0 != (blockSize & 0x07)) maskSize++;
unsigned long blockSpace = sizeof(NormBlock) +
@ -917,6 +933,13 @@ void NormSession::StopSender()
if (flow_control_timer.IsActive())
flow_control_timer.Deactivate();
if (NULL != ack_ex_buffer)
{
delete[] ack_ex_buffer;
ack_ex_buffer = NULL;
ack_ex_length = 0;
}
if (NULL != cmd_buffer)
{
delete[] cmd_buffer;
@ -1045,6 +1068,37 @@ bool NormSession::PreallocateRemoteSender(unsigned int bufferSpace,
return true;
} // end NormSession::PreallocateRemoteSender()
bool NormSession::SetPresetFtiData(unsigned int objectSize,
UINT16 segmentSize,
UINT16 numData,
UINT16 numParity)
{
UINT16 blockSize = numData + numParity;
UINT8 fecM;
if (blockSize > 255)
fecM = 16;
else
fecM = 8;
if (IsSender())
{
if ((segmentSize != segment_size) ||
(numData != ndata) ||
(numParity != nparity) ||
(fecM != fec_m))
{
PLOG(PL_FATAL, "NormSession::SetPresetFtiData() sender FTI mismatch error!\n");
return false;
}
}
preset_fti.SetObjectSize(NormObjectSize(objectSize));
preset_fti.SetSegmentSize(segmentSize);
preset_fti.SetFecMaxBlockLen(numData);
preset_fti.SetFecNumParity(numParity);
preset_fti.SetFecFieldSize(fecM);
preset_fti.SetFecInstanceId(0);
return true;
} // end NormSession::SetPresetFtiData()
void NormSession::Serve()
{
// Only send new data when no other messages are queued for transmission
@ -1341,10 +1395,12 @@ void NormSession::Serve()
}
} // end NormSession::Serve()
void NormSession::SenderSetWatermark(NormObjectId objectId,
bool NormSession::SenderSetWatermark(NormObjectId objectId,
NormBlockId blockId,
NormSegmentId segmentId,
bool overrideFlush)
bool overrideFlush,
const char* appAckReq,
unsigned int appAckReqLen)
{
PLOG(PL_DEBUG, "NormSession::SenderSetWatermark() watermark>%hu:%lu:%hu\n",
(UINT16)objectId, (unsigned long)blockId.GetValue(), (UINT16)segmentId);
@ -1361,7 +1417,42 @@ void NormSession::SenderSetWatermark(NormObjectId objectId,
int robustFactor = GetTxRobustFactor();
while ((next = iterator.GetNextNode()))
static_cast<NormAckingNode*>(next)->Reset(robustFactor);
if (NULL != appAckReq)
{
if (appAckReqLen != ack_ex_length)
{
if (NULL != ack_ex_buffer)
{
delete[] ack_ex_buffer;
ack_ex_buffer = NULL;
ack_ex_length = 0;
}
// Make sure there is room left for at least one acker NormNodeId in
if ((NormHeaderExtension::GetContentOffset() + appAckReqLen) > (segment_size - sizeof(NormNodeId)))
{
PLOG(PL_ERROR, "NormSession::SenderSetWatermark() error: application-defined ACK_REQ content too large!\n");
watermark_pending = false;
return false;
}
else if (NULL == (ack_ex_buffer = new char[appAckReqLen]))
{
PLOG(PL_ERROR, "NormSession::SenderSetWatermark() new app_req_buffer error: %s\n", GetErrorString());
watermark_pending = false;
return false;
}
}
memcpy(ack_ex_buffer, appAckReq, appAckReqLen);
ack_ex_length = appAckReqLen;
}
else if (NULL != ack_ex_buffer)
{
delete[] ack_ex_buffer;
ack_ex_buffer = NULL;
ack_ex_length = 0;
}
PromptSender();
return true;
} // end NormSession::SenderSetWatermark()
void NormSession::SenderResetWatermark()
@ -1504,6 +1595,21 @@ bool NormSession::SenderGetNextAckingNode(NormNodeId& prevNodeId, AckingStatus*
}
} // end NormSession::SenderGetNextAckingNode()
bool NormSession::SenderGetAckEx(NormNodeId nodeId, char* buffer, unsigned int* buflen)
{
NormAckingNode* theNode =
static_cast<NormAckingNode*>(acking_node_tree.FindNodeById(nodeId));
if (NULL != theNode)
{
return theNode->GetAckEx(buffer, buflen);
}
else
{
if (NULL != buflen) *buflen = 0;
return false;
}
} // end NormSession::SenderGetAckEx()
bool NormSession::SenderQueueWatermarkFlush()
{
if (flush_timer.IsActive()) return false;
@ -1511,6 +1617,7 @@ bool NormSession::SenderQueueWatermarkFlush()
if (flush)
{
flush->Init();
flush->SetDestination(address);
flush->SetGrtt(grtt_quantized);
flush->SetBackoffFactor((unsigned char)backoff_factor);
@ -1527,6 +1634,14 @@ bool NormSession::SenderQueueWatermarkFlush()
blockLen = watermark_segment_id;
flush->SetFecPayloadId(fec_id, watermark_block_id.GetValue(), watermark_segment_id, blockLen, fec_m);
if (0 != ack_ex_length)
{
NormAppAckExtension ext;
flush->AttachExtension(ext);
ext.SetContent(ack_ex_buffer, ack_ex_length);
flush->PackExtension(ext);
}
NormNodeTreeIterator iterator(acking_node_tree);
NormAckingNode* next;
watermark_pending = false;
@ -1872,6 +1987,12 @@ bool NormSession::QueueTxObject(NormObject* obj)
return false;
}
if (preset_fti.IsValid() && (obj->GetSize() != preset_fti.GetObjectSize()))
{
PLOG(PL_FATAL, "NormSession::QueueTxObject() preset object info mismatch!\n");
return false;
}
// Manage tx_table min/max count and max size bounds
// Depending on tx cache bounds _and_ what has been
// enqueued/dequeued, we may need to prune the
@ -2424,7 +2545,7 @@ void NormTrace(const struct timeval& currentTime,
"RTT",
"APP"
};
NormMsg::Type msgType = msg.GetType();
UINT16 length = msg.GetLength();
const char* status = sent ? "dst" : "src";
@ -3284,7 +3405,7 @@ void NormSession::SenderHandleCCFeedback(struct timeval currentTime,
// 3) Replace candidate if this response is higher precedence
if (candidate)
{
bool haveRtt = (0 != (ccFlags && NormCC::RTT));
bool haveRtt = (0 != (ccFlags & NormCC::RTT));
bool replace;
if (candidate->GetId() == nodeId)
replace = true;
@ -3377,6 +3498,19 @@ void NormSession::SenderHandleAckMessage(const struct timeval& currentTime, cons
(watermark_block_id == flushAck.GetFecBlockId(fec_m)) &&
(watermark_segment_id == flushAck.GetFecSymbolId(fec_m)))
{
// Cache any application-defined extended ACK content for this acker
NormAppAckExtension ext;
while (ack.GetNextExtension(ext))
{
if (NormHeaderExtension::APP_ACK == ext.GetType())
{
if (!acker->SetAckEx(ext.GetContent(), ext.GetContentLength()))
{
// TBD - notify app of error
PLOG(PL_ERROR, "NormSession::SenderHandleAckMessage() error: unable to cache application-defined ACK content!\n");
}
}
}
acker->MarkAckReceived();
/* This code was an attempt to expedite delivery of the TX_WATERMARK_COMPLETED
notification to the application, but breaks some other desired behavior.
@ -3403,7 +3537,7 @@ void NormSession::SenderHandleAckMessage(const struct timeval& currentTime, cons
{
// This can happen when new watermarks are set when an old watermark is still
// pending (i.e. receivers may still be in the process of replying)
PLOG(PL_DEBUG, "NormSession::SenderHandleAckMessage() received wrong watermark ACK?!\n");
PLOG(PL_DEBUG, "NormSession::SenderHandleAckMessage() received old/wrong watermark ACK?!\n");
}
}
else
@ -4743,6 +4877,7 @@ NormSession::MessageStatus NormSession::SendMessage(NormMsg& msg)
UINT16 msgSize = msg.GetLength();
// Possibly drop some tx messages for testing purposes
bool drop = (tx_loss_rate > 0.0) ? (UniformRand(100.0) < tx_loss_rate) : false;
if (isReceiverMsg && receiver_silent)

2
src/java/jni/normInstanceJni.cpp
View File

@ -182,7 +182,7 @@ JNIEXPORT jobject JNICALL PKGNAME(NormInstance_getNextEvent)
if (env->GetArrayLength(array) <= event.type) {
env->ThrowNew((jclass)env->NewLocalRef(jw_IOException), "Invalid NORM event type (NormEventType.java out of sync with NORM API event header?)");
return false;
return NULL;
}
jobject type = env->GetObjectArrayElement(array, event.type);

55
src/java/src/mil/navy/nrl/norm/enums/NormEventType.java
View File

@ -8,32 +8,33 @@ package mil.navy.nrl.norm.enums;
* @author Jason Rush
*/
public enum NormEventType {
NORM_EVENT_INVALID,
NORM_TX_QUEUE_VACANCY,
NORM_TX_QUEUE_EMPTY,
NORM_TX_FLUSH_COMPLETED,
NORM_TX_WATERMARK_COMPLETED,
NORM_TX_CMD_SENT,
NORM_TX_OBJECT_SENT,
NORM_TX_OBJECT_PURGED,
NORM_TX_RATE_CHANGED,
NORM_LOCAL_SENDER_CLOSED,
NORM_REMOTE_SENDER_NEW,
NORM_REMOTE_SENDER_RESET,
NORM_REMOTE_SENDER_ADDRESS,
NORM_REMOTE_SENDER_ACTIVE,
NORM_REMOTE_SENDER_INACTIVE,
NORM_REMOTE_SENDER_PURGED,
NORM_RX_CMD_NEW,
NORM_RX_OBJECT_NEW,
NORM_RX_OBJECT_INFO,
NORM_RX_OBJECT_UPDATED,
NORM_RX_OBJECT_COMPLETED,
NORM_RX_OBJECT_ABORTED,
NORM_GRTT_UPDATED,
NORM_CC_ACTIVE,
NORM_EVENT_INVALID,
NORM_TX_QUEUE_VACANCY,
NORM_TX_QUEUE_EMPTY,
NORM_TX_FLUSH_COMPLETED,
NORM_TX_WATERMARK_COMPLETED,
NORM_TX_CMD_SENT,
NORM_TX_OBJECT_SENT,
NORM_TX_OBJECT_PURGED,
NORM_TX_RATE_CHANGED,
NORM_LOCAL_SENDER_CLOSED,
NORM_REMOTE_SENDER_NEW,
NORM_REMOTE_SENDER_RESET,
NORM_REMOTE_SENDER_ADDRESS,
NORM_REMOTE_SENDER_ACTIVE,
NORM_REMOTE_SENDER_INACTIVE,
NORM_REMOTE_SENDER_PURGED,
NORM_RX_CMD_NEW,
NORM_RX_OBJECT_NEW,
NORM_RX_OBJECT_INFO,
NORM_RX_OBJECT_UPDATED,
NORM_RX_OBJECT_COMPLETED,
NORM_RX_OBJECT_ABORTED,
NORM_RX_ACK_REQUEST,
NORM_GRTT_UPDATED,
NORM_CC_ACTIVE,
NORM_CC_INACTIVE,
NORM_ACKING_NODE_NEW,
NORM_SEND_ERROR,
NORM_USER_TIMEOUT
NORM_ACKING_NODE_NEW,
NORM_SEND_ERROR,
NORM_USER_TIMEOUT
}

12
src/pynorm/constants.py
View File

@ -69,11 +69,13 @@ NORM_RX_OBJECT_INFO = 18
NORM_RX_OBJECT_UPDATED = 19
NORM_RX_OBJECT_COMPLETED = 20
NORM_RX_OBJECT_ABORTED = 21
NORM_GRTT_UPDATED = 22
NORM_CC_ACTIVE = 23
NORM_CC_INACTIVE = 24
NORM_RX_ACK_REQUEST = 22
NORM_GRTT_UPDATED = 23
NORM_CC_ACTIVE = 24
NORM_CC_INACTIVE = 25
NORM_ACKING_NODE_NEW = 26
NORM_SEND_ERROR = 27
NORM_USER_TIMEOUT = 28
NORM_INSTANCE_INVALID = ctypes.c_void_p.in_dll(libnorm, "NORM_INSTANCE_INVALID").value
NORM_SESSION_INVALID = ctypes.c_void_p.in_dll(libnorm, "NORM_SESSION_INVALID").value

2
src/pynorm/event.py
View File

@ -85,6 +85,8 @@ class Event(object):
return "NORM_RX_OBJECT_COMPLETED"
elif self.type == c.NORM_RX_OBJECT_ABORTED:
return "NORM_RX_OBJECT_ABORTED"
elif self.type == c.NORM_RX_ACK_REQUEST:
return "NORM_RX_ACK_REQUEST"
elif self.type == c.NORM_GRTT_UPDATED:
return "NORM_GRTT_UPDATED"
elif self.type == c.NORM_CC_ACTIVE:

20
wscript
View File

@ -65,8 +65,11 @@ def configure(ctx):
ctx.env.DEFINES_BUILD_NORM += ['NORM_USE_DLL']
if ctx.env.COMPILER_CXX == 'g++' or ctx.env.COMPILER_CXX == 'clang++':
ctx.env.CFLAGS += ['-fvisibility=hidden']
ctx.env.CXXFLAGS += ['-fvisibility=hidden']
ctx.env.CFLAGS += ['-fvisibility=hidden', '-Wno-attributes']
ctx.env.CXXFLAGS += ['-fvisibility=hidden', '-Wno-attributes']
# Will be used by the pkg-config generator
ctx.env.VERSION = VERSION
def build(ctx):
ctx.recurse('protolib')
@ -132,7 +135,8 @@ def build(ctx):
if ctx.env.BUILD_JAVA:
ctx.shlib(
target = 'mil_navy_nrl_norm',
use = ['norm_shlib', 'JAVA'],
includes = ['include'],
use = ['norm', 'JAVA'],
vnum = VERSION,
source = ['src/java/jni/{0}.cpp'.format(x) for x in [
'normJni',
@ -205,6 +209,16 @@ def build(ctx):
# Enable example targets specified on the command line
ctx._parse_targets()
# Generate pkg-config file
# Add additional static compilation dependencies based on the system.
# libpcap is used by protolib on GNU/Hurd based systems.
static_libs = ''
if ctx.options.enable_static_library:
static_libs += ' -lstdc++ -lprotolib'
if system == "gnu":
static_libs += ' -lpcap'
ctx(source='norm.pc.in', STATIC_LIBS = static_libs)
def _make_simple_example(ctx, name, path='examples'):
'''Makes a task from a single source file in the examples directory.