diff --git a/README.md b/README.md index 48020f3..980a03a 100644 --- a/README.md +++ b/README.md @@ -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: - + 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 diff --git a/VERSION.TXT b/VERSION.TXT index b6e6f76..74874d6 100644 --- a/VERSION.TXT +++ b/VERSION.TXT @@ -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 ============= diff --git a/android/README.TXT b/android/README.TXT new file mode 100644 index 0000000..4d5a342 --- /dev/null +++ b/android/README.TXT @@ -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. + + + + diff --git a/android/build.gradle b/android/build.gradle new file mode 100644 index 0000000..651d63d --- /dev/null +++ b/android/build.gradle @@ -0,0 +1,16 @@ +buildscript { + repositories { + google() + jcenter() + } + dependencies { + classpath 'com.android.tools.build:gradle:3.1.1' + } +} + +allprojects { + repositories { + google() + jcenter() + } +} diff --git a/android/gradle/wrapper/LICENSE-for-gradlewrapper.txt b/android/gradle/wrapper/LICENSE-for-gradlewrapper.txt new file mode 100644 index 0000000..e06d208 --- /dev/null +++ b/android/gradle/wrapper/LICENSE-for-gradlewrapper.txt @@ -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. + diff --git a/android/gradle/wrapper/gradle-wrapper.jar b/android/gradle/wrapper/gradle-wrapper.jar new file mode 100644 index 0000000..13372ae Binary files /dev/null and b/android/gradle/wrapper/gradle-wrapper.jar differ diff --git a/android/gradle/wrapper/gradle-wrapper.properties b/android/gradle/wrapper/gradle-wrapper.properties new file mode 100644 index 0000000..d384d28 --- /dev/null +++ b/android/gradle/wrapper/gradle-wrapper.properties @@ -0,0 +1 @@ +distributionUrl=https\://services.gradle.org/distributions/gradle-4.4-all.zip \ No newline at end of file diff --git a/android/gradlew b/android/gradlew new file mode 100755 index 0000000..9d82f78 --- /dev/null +++ b/android/gradlew @@ -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 "$@" diff --git a/android/gradlew.bat b/android/gradlew.bat new file mode 100644 index 0000000..aec9973 --- /dev/null +++ b/android/gradlew.bat @@ -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 diff --git a/android/settings.gradle b/android/settings.gradle new file mode 100644 index 0000000..2b49d90 --- /dev/null +++ b/android/settings.gradle @@ -0,0 +1 @@ +include ':lib' \ No newline at end of file diff --git a/examples/normStreamer.cpp b/examples/normStreamer.cpp index bcc23ae..81c2604 100644 --- a/examples/normStreamer.cpp +++ b/examples/normStreamer.cpp @@ -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(¤tTime, 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 {send|recv} [addr [/]][interface ][loopback]\n" + fprintf(stderr, "Usage: normStreamer id {send|recv} [addr [/]][interface ][loopback][info]\n" " [cc|cce|ccl|rate ][ack [,,...][flush {none|passive|active}]\n" " [listen [/]][linterface ]\n" " [relay /][limit [/]][output ]\n" @@ -1244,6 +1332,7 @@ void PrintHelp() " interface -- 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 -- 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 [] -- 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", diff --git a/include/normApi.h b/include/normApi.h index 3beaae0..412e20b 100644 --- a/include/normApi.h +++ b/include/normApi.h @@ -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); diff --git a/include/normMessage.h b/include/normMessage.h index c1bf7be..ae68765 100644 --- a/include/normMessage.h +++ b/include/normMessage.h @@ -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; diff --git a/include/normNode.h b/include/normNode.h index 320c712..cc24c5a 100644 --- a/include/normNode.h +++ b/include/normNode.h @@ -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; diff --git a/include/normObject.h b/include/normObject.h index 7b366e9..7f0212f 100644 --- a/include/normObject.h +++ b/include/normObject.h @@ -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);} diff --git a/include/normSession.h b/include/normSession.h index 379c4c5..03f076f 100644 --- a/include/normSession.h +++ b/include/normSession.h @@ -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; diff --git a/include/normVersion.h b/include/normVersion.h index 4e969fa..4be5779 100644 --- a/include/normVersion.h +++ b/include/normVersion.h @@ -36,6 +36,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.5.8" +#define VERSION "1.5.9" #endif // _NORM_VERSION diff --git a/makefiles/Makefile.common b/makefiles/Makefile.common index e1bdf27..5c698d3 100644 --- a/makefiles/Makefile.common +++ b/makefiles/Makefile.common @@ -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) diff --git a/makefiles/android/AndroidManifest.xml b/makefiles/android-ndk/AndroidManifest.xml similarity index 100% rename from makefiles/android/AndroidManifest.xml rename to makefiles/android-ndk/AndroidManifest.xml diff --git a/makefiles/android/README b/makefiles/android-ndk/README similarity index 85% rename from makefiles/android/README rename to makefiles/android-ndk/README index 738fd34..47e7fb4 100644 --- a/makefiles/android/README +++ b/makefiles/android-ndk/README @@ -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. diff --git a/makefiles/android/ant.properties b/makefiles/android-ndk/ant.properties similarity index 100% rename from makefiles/android/ant.properties rename to makefiles/android-ndk/ant.properties diff --git a/makefiles/android/build.xml b/makefiles/android-ndk/build.xml similarity index 100% rename from makefiles/android/build.xml rename to makefiles/android-ndk/build.xml diff --git a/makefiles/android/jni/Android.mk b/makefiles/android-ndk/jni/Android.mk similarity index 96% rename from makefiles/android/jni/Android.mk rename to makefiles/android-ndk/jni/Android.mk index 6332712..319e077 100644 --- a/makefiles/android/jni/Android.mk +++ b/makefiles/android-ndk/jni/Android.mk @@ -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) diff --git a/makefiles/android/jni/Application.mk b/makefiles/android-ndk/jni/Application.mk similarity index 100% rename from makefiles/android/jni/Application.mk rename to makefiles/android-ndk/jni/Application.mk diff --git a/makefiles/android/proguard-project.txt b/makefiles/android-ndk/proguard-project.txt similarity index 100% rename from makefiles/android/proguard-project.txt rename to makefiles/android-ndk/proguard-project.txt diff --git a/makefiles/android/project.properties b/makefiles/android-ndk/project.properties similarity index 100% rename from makefiles/android/project.properties rename to makefiles/android-ndk/project.properties diff --git a/norm.pc.in b/norm.pc.in new file mode 100644 index 0000000..d824ab6 --- /dev/null +++ b/norm.pc.in @@ -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} diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index ff510aa..7379d96 100644 --- a/src/common/normApi.cpp +++ b/src/common/normApi.cpp @@ -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(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(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(node); + result = sender->GetWatermarkEx(buffer, buflen); + } + instance->dispatcher.ResumeThread(); + } + } + return result; +} // end NormNodeGetWatermarkEx() + NORM_API_LINKAGE void NormNodeFreeBuffers(NormNodeHandle nodeHandle) { diff --git a/src/common/normApp.cpp b/src/common/normApp.cpp index 3a9c527..98f9bd3 100644 --- a/src/common/normApp.cpp +++ b/src/common/normApp.cpp @@ -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* 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() diff --git a/src/common/normEncoderRS16.cpp b/src/common/normEncoderRS16.cpp index 9788452..1aba353 100644 --- a/src/common/normEncoderRS16.cpp +++ b/src/common/normEncoderRS16.cpp @@ -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) ; } } diff --git a/src/common/normMessage.cpp b/src/common/normMessage.cpp index 1a474b2..900ecaf 100644 --- a/src/common/normMessage.cpp +++ b/src/common/normMessage.cpp @@ -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: diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp index 505471f..2393a1d 100644 --- a/src/common/normNode.cpp +++ b/src/common/normNode.cpp @@ -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(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(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(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(preset_stream)))) { obj = static_cast(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; diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index 416c5dd..96d6226 100644 --- a/src/common/normObject.cpp +++ b/src/common/normObject.cpp @@ -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(); diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index 56161ed..490899c 100644 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -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(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(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) diff --git a/src/java/jni/normInstanceJni.cpp b/src/java/jni/normInstanceJni.cpp index aaef916..028b08e 100644 --- a/src/java/jni/normInstanceJni.cpp +++ b/src/java/jni/normInstanceJni.cpp @@ -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); diff --git a/src/java/src/mil/navy/nrl/norm/enums/NormEventType.java b/src/java/src/mil/navy/nrl/norm/enums/NormEventType.java index 8ba2f56..64ee2fb 100644 --- a/src/java/src/mil/navy/nrl/norm/enums/NormEventType.java +++ b/src/java/src/mil/navy/nrl/norm/enums/NormEventType.java @@ -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 } diff --git a/src/pynorm/constants.py b/src/pynorm/constants.py index c2ad5d0..f4d06d3 100644 --- a/src/pynorm/constants.py +++ b/src/pynorm/constants.py @@ -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 diff --git a/src/pynorm/event.py b/src/pynorm/event.py index c3b1ba9..3db3f7b 100644 --- a/src/pynorm/event.py +++ b/src/pynorm/event.py @@ -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: diff --git a/wscript b/wscript index 8428899..b17bebc 100644 --- a/wscript +++ b/wscript @@ -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.