added option for GRTT probing aspect of NORM to use a distinct IP traffic class for its operation. The reason for this is where some Layer 2 protocols may implement some form of reliable, retransmission that could adversely impact the NORM GRTT round-trip measurement. A distinct traffic class for that signaling allows the possibility of mapping it to different underlying Layer 2 services (e.g. bypass any reliability mechanism for those specific NORM messages)

examples/normCast.cpp

@@ -144,6 +144,10 @@ class NormCaster
         }
         void SetSilentReceiver(bool state)
             {NormSetSilentReceiver(norm_session, true);}
+        
+        void SetProbeTOS(UINT8 value)
+            {probe_tos = value;}
+        
         void SetTxLoss(double txloss)
             {NormSetTxLoss(norm_session, txloss);}
             
@@ -156,6 +160,7 @@ class NormCaster
         unsigned int                        tx_pending_prefix_len;
         bool                                is_multicast;
         bool                                loopback;
+        UINT8                               probe_tos;
         unsigned int                        norm_tx_segment_size;
         unsigned int                        norm_tx_queue_max;   // max number of objects that can be enqueued at once 
         unsigned int                        norm_tx_queue_count; // count of unacknowledged enqueued objects (TBD - optionally track size too)
@@ -178,7 +183,7 @@ class NormCaster
 NormCaster::NormCaster()
  : norm_session(NORM_SESSION_INVALID), tx_file_iterator(tx_file_list), 
    tx_pending_prefix_len(0), 
-   is_multicast(false), loopback(false), norm_tx_segment_size(1400), 
+   is_multicast(false), loopback(false), probe_tos(0), norm_tx_segment_size(1400), 
    norm_tx_queue_max(8), norm_tx_queue_count(0), 
    norm_flow_control_pending(false), norm_tx_vacancy(true), norm_acking(false), 
    norm_flushing(true), norm_flush_object(NORM_OBJECT_INVALID), norm_last_object(NORM_OBJECT_INVALID),
@@ -243,6 +248,8 @@ bool NormCaster::OpenNormSession(NormInstanceHandle instance, const char* addr,
     
     //NormSetTxRobustFactor(norm_session, 20);
     
+    NormSetGrttProbingTOS(norm_session, probe_tos);
+    
     return true;
 }  // end NormCaster::OpenNormSession()
 
@@ -662,7 +669,7 @@ void Usage()
 {
     fprintf(stderr, "Usage: normCast id <nodeId> {send <file/dir list> &| recv <rxCacheDir>}\n"
                     "                [addr <addr>[/<port>]][interface <name>][ack <node1>[,<node2>,...]\n"
-                    "                [cc|cce|ccl|rate <bitsPerSecond>]\n"
+                    "                [cc|cce|ccl|rate <bitsPerSecond>][ptos <value>]\n"
                     "                [debug <level>][trace]\n");
 }  // end Usage()
 
@@ -746,6 +753,27 @@ int main(int argc, char* argv[])
         {
             loopback = true;
         }
+        else if (0 == strncmp(cmd, "ptos", len))
+        {
+            if (i >= argc)
+            {
+                fprintf(stderr, "normStreamer error: missing 'ptos' value!\n");
+                Usage();
+                return -1;
+            }
+            int tos = -1;
+            int result = sscanf(argv[i], "%i", &tos);
+            if (1 != result)
+                result = sscanf(argv[i], "%x", &tos);
+            if ((1 != result) || (tos < 0) || (tos > 255))
+            {
+                fprintf(stderr, "normStreamer error: invalid 'ptos' value!\n");
+                Usage();
+                return -1;
+            }
+            i++;
+            normCast.SetProbeTOS(tos);
+        }
         else if (0 == strncmp(cmd, "addr", len))
         {
             if (i >= argc)

examples/normStreamer.cpp

@@ -279,6 +279,9 @@ class NormStreamer
         void SetOutputSocketBufferSize(unsigned int value)
             {output_socket_buffer_size = value;}
         
+        void SetProbeTOS(UINT8 value)
+            {probe_tos = value;}
+        
         // Check that sequence numbers increase by one each time.
     	// Assumes that sequence number is 8- or 4-byte network-order first 8 bytes of buffer.
     	void CheckSequenceNumber(const char* buffer, const char* source);
@@ -290,6 +293,7 @@ class NormStreamer
     private:
         NormSessionHandle   norm_session;
         bool                is_multicast;
+        bool                probe_tos;
         bool                loopback;
         bool                is_running;     
                                                      
@@ -359,7 +363,7 @@ class NormStreamer
 };  // end class NormStreamer
 
 NormStreamer::NormStreamer()
- : norm_session(NORM_SESSION_INVALID), is_multicast(false), loopback(false), is_running(false),
+ : norm_session(NORM_SESSION_INVALID), is_multicast(false), probe_tos(0), loopback(false), is_running(false),
    input_socket(ProtoSocket::UDP), input_file(stdin), input_fd(fileno(stdin)), input_ready(true), 
    input_needed(false), input_msg_length(0), input_index(0),
    tx_stream (NORM_OBJECT_INVALID), tx_ready(true),
@@ -570,6 +574,8 @@ bool NormStreamer::OpenNormSession(NormInstanceHandle instance, const char* addr
     
     NormSetTxRobustFactor(norm_session, 20);
     
+    NormSetGrttProbingTOS(norm_session, probe_tos);
+    
     return true;
 }  // end NormStreamer::OpenNormSession()
 
@@ -1338,6 +1344,7 @@ void PrintHelp()
 			"   interface <name>        -- Specifies the name of the network interface on which to conduct NORM protocol\n"
 			"                              (e.g., 'eth0')\n"
             "   loopback                -- Enables 'loopback' sessions on the same host machine.  Required for multicast loopback.\n"
+            "   ptos <value>            -- Set special IP traffic class (TOS) for GRTT probing and acknowledgments\n"
             "   info                    -- Limits FTI header extension to NORM_INFO message only (reduced overhead)\n"
             "   rate <bitsPerSecond>    -- sets fixed sender rate (and receiver token bucket rate if 'limit' option is used)\n"
             "   [cc|cce|ccl]            -- Enables optional NORM congestion control mode (overrides 'rate')\n"
@@ -1441,6 +1448,27 @@ int main(int argc, char* argv[])
         {
             loopback = true;
         }
+        else if (0 == strncmp(cmd, "ptos", len))
+        {
+            if (i >= argc)
+            {
+                fprintf(stderr, "normStreamer error: missing 'ptos' value!\n");
+                Usage();
+                return -1;
+            }
+            int tos = -1;
+            int result = sscanf(argv[i], "%i", &tos);
+            if (1 != result)
+                result = sscanf(argv[i], "%x", &tos);
+            if ((1 != result) || (tos < 0) || (tos > 255))
+            {
+                fprintf(stderr, "normStreamer error: invalid 'ptos' value!\n");
+                Usage();
+                return -1;
+            }
+            i++;
+            normStreamer.SetProbeTOS(tos);
+        }
         else if (0 == strncmp(cmd, "info", len))
         {
             ftiInfo = true;

include/normApi.h

@@ -515,6 +515,10 @@ void NormSetGrttProbingInterval(NormSessionHandle sessionHandle,
                                 double            intervalMin,
                                 double            intervalMax);
 
+NORM_API_LINKAGE
+void NormSetGrttProbingTOS(NormSessionHandle sessionHandle,
+                           UINT8              probeTOS);
+
 NORM_API_LINKAGE 
 void NormSetBackoffFactor(NormSessionHandle sessionHandle,
                           double            backoffFactor);

include/normMessage.h

@@ -656,6 +656,7 @@ class NormMsg
         
         // To retrieve any attached header extensions
         bool HasExtensions() const {return (header_length > header_length_base);}
+        bool HasExtension(NormHeaderExtension::Type extType);
         bool GetNextExtension(NormHeaderExtension& ext) const
         {
             const UINT32* currentBuffer = ext.GetBuffer();
@@ -1989,7 +1990,7 @@ class NormNackMsg : public NormMsg
         };    
 };  // end class NormNackMsg
 
-class NormAckMsg : public NormMsg
+class NormAckMsg : public NormAck, public NormMsg
 {
     public:
         

include/normObject.h

@@ -52,10 +52,7 @@ class NormObject
         
         // This must be reset after each update
         void SetNotifyOnUpdate(bool state)
-        {
+            {notify_on_update = state;}
-                TRACE("enter SetNotifyOnUpdate(%d) ...\n", state);
-                notify_on_update = state;
-        }
         
         // Object information
         NormObject::Type GetType() const {return type;}

include/normSession.h

@@ -289,6 +289,12 @@ class NormSession
             if (state) probe_proactive = true;
         }
         
+        // This MUST be called before
+        void SetProbeTOS(UINT8 probeTOS)
+            {probe_tos = probeTOS;}
+        UINT8 GetProbeTOS() const
+            {return probe_tos;}
+        
         // This method enables/disables flow control operation.
         void SetFlowControl(double flowControlFactor)  
             {flow_control_factor = flowControlFactor;}
@@ -674,10 +680,15 @@ class NormSession
         void TxSocketRecvHandler(ProtoSocket& theSocket, ProtoSocket::Event theEvent);
         void RxSocketRecvHandler(ProtoSocket& theSocket, ProtoSocket::Event theEvent);        
         void HandleReceiveMessage(NormMsg& msg, bool wasUnicast, bool ecn = false);
-        
+
+#ifdef ECN_SUPPORT        
         // This is used when raw packet capture is enabled
+        bool OpenProtoCap();
+        void CloseProtoCap();
+        bool RawSendTo(const char* buffer, unsigned int& numBytes, const ProtoAddress& dstAddr, UINT8 trafficClass);
         void OnPktCapture(ProtoChannel&              theChannel,
 	                      ProtoChannel::Notification notifyType);
+#endif // ECN_SUPPORT
         
         // Sender message handling routines
         void SenderHandleNackMessage(const struct timeval& currentTime, 
@@ -723,7 +734,10 @@ class NormSession
         ProtoSocket                     tx_socket_actual;
         ProtoSocket*                    tx_socket;
         ProtoSocket                     rx_socket;
-        ProtoCap*                       rx_cap;        // raw packet capture alternative to "rx_socket"
+#ifdef ECN_SUPPORT
+        ProtoCap*                       proto_cap;        // raw packet capture alternative to "rx_socket"
+        ProtoAddress                    src_addr;         // used for raw packet sendto()
+#endif // ECN_SUPPORT
         bool                            rx_port_reuse; // enable rx_socket port (sessionPort) reuse when true
         ProtoAddress                    rx_bind_addr;
         ProtoAddress                    rx_connect_addr;
@@ -817,6 +831,7 @@ class NormSession
         bool                            probe_reset;   
         bool                            probe_data_check;  // refrain cc probe until data is send
         struct timeval                  probe_time_last;
+        UINT8                           probe_tos;         // optionally use different IP TOS for GRTT probe/response
         
         double                          grtt_interval;     // current GRTT update interval
         double                          grtt_interval_min; // minimum GRTT update interval

src/common/normApi.cpp

@@ -1757,6 +1757,19 @@ void NormSetGrttProbingInterval(NormSessionHandle sessionHandle,
     }
 }  // end NormSetGrttProbingInterval()
 
+NORM_API_LINKAGE
+void NormSetGrttProbingTOS(NormSessionHandle sessionHandle,
+                           UINT8              probeTOS)
+{
+    NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
+    if (instance && instance->dispatcher.SuspendThread())
+    {
+        NormSession* session = (NormSession*)sessionHandle;
+        session->SetProbeTOS(probeTOS);
+        instance->dispatcher.ResumeThread();
+    }
+}  // end NormSetGrttProbingTOS()
+
 NORM_API_LINKAGE
 void NormSetBackoffFactor(NormSessionHandle sessionHandle,
                           double            backoffFactor)

src/common/normFile.cpp

@@ -193,7 +193,7 @@ void NormFile::Unlock()
 #ifdef HAVE_FLOCK
     if (0 != flock(fd, LOCK_UN))
     {
-        PLOG(PL_ERROR, "NormFile::Unlock() flock() error: %s\n", GetErrorString());
+        PLOG(PL_ERROR, "NormFile::Unlock() flock(%d) error: %s\n", fd, GetErrorString());
     }
 #else
 #ifdef HAVE_LOCKF

src/common/normMessage.cpp

@@ -98,6 +98,18 @@ void NormMsg::Display() const
         PLOG(PL_ALWAYS, "%02x", *ptr++);
 }  // end NormMsg::Display()
 
+bool NormMsg::HasExtension(NormHeaderExtension::Type extType)
+{
+    
+    NormHeaderExtension ext;
+    while (GetNextExtension(ext))
+    {
+        if (ext.GetType() == extType)
+            return true;
+    }
+    return false;
+}  // end NormMsg::HasExtension()
+
 bool NormCmdCCMsg::GetCCNode(NormNodeId     nodeId, 
                              UINT8&         flags, 
                              UINT8&         rtt, 

src/common/normObject.cpp

@@ -2315,7 +2315,8 @@ bool NormFileObject::Open(const char* thePath,
         {
             if (file.Open(thePath, O_RDWR | O_CREAT | O_TRUNC))
             {
-                file.Lock();   
+                if (!file.Lock())
+                    PLOG(PL_WARN, "NormFileObject::Open() warning: NormFile::Lock() failure\n");
             }   
             else
             {
@@ -2392,10 +2393,13 @@ bool NormFileObject::Accept(const char* thePath)
 
 void NormFileObject::Close()
 {
+    if (file.IsOpen())
+    {
+        if (NULL != sender)  // we've been receiving this file
+            file.Unlock();
+        file.Close();
+    }
     NormObject::Close();
-    if (NULL != sender)  // we've been receiving this file
-        file.Unlock();
-    file.Close();
 }  // end NormFileObject::Close()
 
 bool NormFileObject::WriteSegment(NormBlockId   blockId, 

src/common/normSession.cpp

@@ -8,6 +8,7 @@
 
 #include "protoPktETH.h"
 #include "protoPktIP.h"
+#include "protoNet.h"
 
 const UINT8 NormSession::DEFAULT_TTL = 255;
 const double NormSession::DEFAULT_TRANSMIT_RATE = 64000.0;  // bits/sec
@@ -44,7 +45,7 @@ enum NormSocketCommand
 NormSession::NormSession(NormSessionMgr &sessionMgr, NormNodeId localNodeId)
     : session_mgr(sessionMgr), notify_pending(false), tx_port(0), tx_port_reuse(false),
       tx_socket_actual(ProtoSocket::UDP), tx_socket(&tx_socket_actual),
-      rx_socket(ProtoSocket::UDP), rx_cap(NULL), rx_port_reuse(false), local_node_id(localNodeId),
+      rx_socket(ProtoSocket::UDP), proto_cap(NULL), rx_port_reuse(false), local_node_id(localNodeId),
       ttl(DEFAULT_TTL), tos(0), loopback(false), mcast_loopback(false), fragmentation(false), ecn_enabled(false),
       tx_rate(DEFAULT_TRANSMIT_RATE / 8.0), tx_rate_min(-1.0), tx_rate_max(-1.0), tx_residual(0),
       backoff_factor(DEFAULT_BACKOFF_FACTOR), is_sender(false),
@@ -60,8 +61,7 @@ NormSession::NormSession(NormSessionMgr &sessionMgr, NormNodeId localNodeId)
       tx_repair_pending(false), advertise_repairs(false),
       suppress_nonconfirmed(false), suppress_rate(-1.0), suppress_rtt(-1.0),
       probe_proactive(true), probe_pending(false), probe_reset(true), probe_data_check(false),
-      grtt_interval(0.5),
+      probe_tos(0), grtt_interval(0.5), grtt_interval_min(DEFAULT_GRTT_INTERVAL_MIN),
-      grtt_interval_min(DEFAULT_GRTT_INTERVAL_MIN),
       grtt_interval_max(DEFAULT_GRTT_INTERVAL_MAX),
       grtt_max(DEFAULT_GRTT_MAX),
       grtt_decrease_delay_count(DEFAULT_GRTT_DECREASE_DELAY),
@@ -189,6 +189,8 @@ bool NormSession::Open()
         {
             tx_socket = &rx_socket;
         }
+        tx_port = tx_socket->GetPort();
+            
     }
     if (!rx_socket.IsOpen() && (!tx_only || (&rx_socket == tx_socket)))
     {
@@ -298,61 +300,23 @@ bool NormSession::Open()
 #ifdef ECN_SUPPORT
     // TBD - do this via UDP socket recvmsg() instead of raw packet capture
     // If raw packet capture is enabled, create/open ProtoCap device to do it
-    if (ecn_enabled && !tx_only)
+    if ((ecn_enabled && !tx_only) || (0 != probe_tos))
     {
-        if (NULL == rx_cap)
+        if (!OpenProtoCap())
         {
-            if (NULL == (rx_cap = ProtoCap::Create()))
+            PLOG(PL_FATAL, "NormSession::Open() error: unable to create ProtoCap device!\n");
-            {
+            Close();
-                PLOG(PL_FATAL, "NormSession::Open() error: unable to create ProtoCap device!\n");
+            return false;
-                Close();
+        }
-                return false;
+        if (ecn_enabled)
-            }
+        {
-            rx_cap->SetListener(this, &NormSession::OnPktCapture);
+            rx_socket.StopInputNotification(); // Disable rx_socket (keep open so mcast JOIN holds)
-            rx_cap->SetNotifier(session_mgr.GetChannelNotifier());
+            proto_cap->StartInputNotification();
-            if (!rx_cap->Open(('\0' != interface_name[0]) ? interface_name : NULL))
+        }
-            {
+        else
-                PLOG(PL_FATAL, "NormSession::Open() error: unable to open ProtoCap device '%s'!\n", (('\0' != interface_name[0]) ? interface_name : "(null)"));
+        {
-                Close();
+            proto_cap->StopInputNotification();
-                return false;
-            }
-            rx_cap->StartInputNotification();
-            // Populate "dst_addr_list" with potential valid dst addrs for this host
-            dst_addr_list.Destroy();
-            if (rx_bind_addr.IsValid())
-            {
-                if (!dst_addr_list.Insert(rx_bind_addr))
-                {
-                    PLOG(PL_FATAL, "NormSession::Open() error: unable to add rx_bind_addr to dst_addr_list!!\n");
-                    Close();
-                    return false;
-                }
-            }
-            else
-            {
-                // Put all local unicast addrs in list
-                if (!ProtoSocket::GetHostAddressList(ProtoAddress::IPv4, dst_addr_list))
-                    PLOG(PL_WARN, "NormSession::Open() warning: incomplete IPv4 host address list\n");
-                if (!ProtoSocket::GetHostAddressList(ProtoAddress::IPv6, dst_addr_list))
-                    PLOG(PL_WARN, "NormSession::Open() warning: incomplete IPv6 host address list\n");
-            }
-            if (address.IsMulticast() && !address.HostIsEqual(rx_bind_addr))
-            {
-                if (!dst_addr_list.Insert(address))
-                {
-                    PLOG(PL_FATAL, "NormSession::Open() error: unable to add session addr to dst_addr_list!!\n");
-                    Close();
-                    return false;
-                }
-            }
-            if (dst_addr_list.IsEmpty())
-            {
-                PLOG(PL_FATAL, "NormSession::Open() error: unable to add any addresses to dst_addr_list!!\n");
-                Close();
-                return false;
-            }
         }
-        rx_socket.StopInputNotification(); // Disable rx_socket (keep open so mcast JOIN holds)
     }
 #endif // ECN_SUPPORT
     if (message_pool.IsEmpty())
@@ -402,15 +366,85 @@ void NormSession::Close()
         rx_socket.Close();
     }
 #ifdef ECN_SUPPORT
-    if (NULL != rx_cap)
+    CloseProtoCap();
-    {
-        rx_cap->Close();
-        delete rx_cap;
-        rx_cap = NULL;
-    }
 #endif // ECN_SUPPORT
 } // end NormSession::Close()
 
+#ifdef ECN_SUPPORT
+bool NormSession::OpenProtoCap()
+{
+    if (NULL == proto_cap)
+    {
+        if (NULL == (proto_cap = ProtoCap::Create()))
+        {
+            PLOG(PL_FATAL, "NormSession::OpenProtoCap() error: unable to create ProtoCap device!\n");
+            return false;
+        }
+        proto_cap->SetListener(this, &NormSession::OnPktCapture);
+        proto_cap->SetNotifier(session_mgr.GetChannelNotifier());
+        if (!proto_cap->Open(('\0' != interface_name[0]) ? interface_name : NULL))
+        {
+            PLOG(PL_FATAL, "NormSession::OpenProtoCap() error: unable to open ProtoCap device '%s'!\n", (('\0' != interface_name[0]) ? interface_name : "(null)"));
+            return false;
+        }
+        // Populate "dst_addr_list" with potential valid dst addrs for this host
+        dst_addr_list.Destroy();
+        if (rx_bind_addr.IsValid())
+        {
+            if (!dst_addr_list.Insert(rx_bind_addr))
+            {
+                PLOG(PL_FATAL, "NormSession::OpenProtoCap() error: unable to add rx_bind_addr to dst_addr_list!!\n");
+                return false;
+            }
+        }
+        else
+        {
+            // Put all local unicast addrs in list
+            if (!ProtoSocket::GetHostAddressList(ProtoAddress::IPv4, dst_addr_list))
+                PLOG(PL_WARN, "NormSession::OpenProtoCap() warning: incomplete IPv4 host address list\n");
+            if (!ProtoSocket::GetHostAddressList(ProtoAddress::IPv6, dst_addr_list))
+                PLOG(PL_WARN, "NormSession::OpenProtoCap() warning: incomplete IPv6 host address list\n");
+        }
+        if (address.IsMulticast() && !address.HostIsEqual(rx_bind_addr))
+        {
+            if (!dst_addr_list.Insert(address))
+            {
+                PLOG(PL_FATAL, "NormSession::OpenProtoCap() error: unable to add session addr to dst_addr_list!!\n");
+                return false;
+            }
+        }
+        
+        // Get the interface source address for pcap message transmission
+        if (!ProtoNet::GetInterfaceAddress(proto_cap->GetInterfaceIndex(),
+                                            address.GetType(),
+                                            src_addr))
+        {
+            PLOG(PL_WARN, "NormSession::OpenProtoCap() warning: unable to get interface source address\n");
+        }
+        
+        TRACE("got source address: %s\n", src_addr.GetHostString());
+        
+        if (dst_addr_list.IsEmpty())
+        {
+            PLOG(PL_FATAL, "NormSession::OpenProtoCap() error: unable to add any addresses to dst_addr_list!!\n");
+            return false;
+        }
+    }
+    return true;
+}  // end NormSession::OpenProtoCap()
+
+void NormSession::CloseProtoCap()
+{
+    if (NULL != proto_cap)
+    {
+        proto_cap->Close();
+        delete proto_cap;
+        proto_cap = NULL;
+    }
+}  // end NormSession::CloseProtoCap()
+
+#endif  // ECN_SUPPORT
+
 bool NormSession::SetMulticastInterface(const char *interfaceName)
 {
     if (NULL != interfaceName)
@@ -548,12 +582,8 @@ void NormSession::SetTxOnly(bool txOnly, bool connectToSessionAddress)
             if (rx_socket.IsOpen())
                 rx_socket.Close();
 #ifdef ECN_SUPPORT
-            if (NULL != rx_cap)
+            if (NULL != proto_cap)
-            {
+                proto_cap->StopInputNotification();
-                rx_cap->Close();
-                delete rx_cap;
-                rx_cap = NULL;
-            }
 #endif // ECN_SUPPORT
         }
         // We connect tx_only session sockets when tx port
@@ -2408,6 +2438,7 @@ void NormSession::RxSocketRecvHandler(ProtoSocket &theSocket,
     } // end if/else (theEvent == RECV/SEND)
 } // end NormSession::RxSocketRecvHandler()
 
+#ifdef ECN_SUPPORT
 #ifndef SIMULATE
 void NormSession::OnPktCapture(ProtoChannel &theChannel,
                                ProtoChannel::Notification notifyType)
@@ -2539,6 +2570,7 @@ void NormSession::OnPktCapture(ProtoChannel &theChannel,
     } // end while(1)
 } // end NormSession::OnPktCapture()
 #endif // !SIMULATE
+#endif // ECN_SUPPORT
 
 // TBD - move this to its own cpp file???
 void NormTrace(const struct timeval &currentTime,
@@ -4841,6 +4873,7 @@ NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg)
     //TRACE("sending message length %hu\n", msg.GetLength());
     bool isReceiverMsg = false;
     bool isProbe = false;
+    bool sendRaw = false;
 
     // Fill in any last minute timestamps
     // (TBD) fill in InstanceId fields on all messages as needed
@@ -4849,80 +4882,98 @@ NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg)
     UINT16 instId = instance_id; // assume it's a sender message (will be overridden otherwise)
     switch (msg.GetType())
     {
-    case NormMsg::INFO:
+        case NormMsg::INFO:
-    case NormMsg::DATA:
+        case NormMsg::DATA:
-    {
-        NormObjectMsg &objMsg = static_cast<NormObjectMsg &>(msg);
-        objMsg.SetInstanceId(instId);
-        msg.SetSequence(tx_sequence++); // (TBD) set for session dst msgs
-        if (syn_status)
-            objMsg.SetFlag(NormObjectMsg::FLAG_SYN);
-        break;
-    }
-    case NormMsg::CMD:
-    {
-        NormCmdMsg &cmd = static_cast<NormCmdMsg &>(msg);
-        ((NormCmdMsg &)msg).SetInstanceId(instId);
-        switch (cmd.GetFlavor())
         {
-        case NormCmdMsg::CC:
+            NormObjectMsg &objMsg = static_cast<NormObjectMsg &>(msg);
-        {
+            objMsg.SetInstanceId(instId);
-            NormCmdCCMsg &ccMsg = static_cast<NormCmdCCMsg &>(cmd);
+            msg.SetSequence(tx_sequence++); // (TBD) set for session dst msgs
-            struct timeval currentTime;
-            ProtoSystemTime(currentTime);
-            ccMsg.SetSendTime(currentTime);
-            isProbe = true;
             if (syn_status)
-                ccMsg.SetSyn();
+                objMsg.SetFlag(NormObjectMsg::FLAG_SYN);
             break;
         }
-        case NormCmdMsg::SQUELCH:
+        case NormMsg::CMD:
+        {
+            NormCmdMsg &cmd = static_cast<NormCmdMsg &>(msg);
+            ((NormCmdMsg &)msg).SetInstanceId(instId);
+            switch (cmd.GetFlavor())
+            {
+                case NormCmdMsg::CC:
+                {
+                    NormCmdCCMsg &ccMsg = static_cast<NormCmdCCMsg &>(cmd);
+                    struct timeval currentTime;
+                    ProtoSystemTime(currentTime);
+                    ccMsg.SetSendTime(currentTime);
+                    isProbe = true;
+                    if (0 != probe_tos)
+                        sendRaw = true;  // so probe will be marked accordingly
+                    if (syn_status)
+                        ccMsg.SetSyn();
+                    break;
+                }
+                case NormCmdMsg::SQUELCH:
+                    break;
+                default:
+                    break;
+            }
+            msg.SetSequence(tx_sequence++); // (TBD) set for session dst msgs
             break;
+        }
+        case NormMsg::NACK:
+        {
+            msg.SetSequence(0); // TBD - set per destination
+            isReceiverMsg = true;
+            NormNackMsg &nack = (NormNackMsg &)msg;
+            NormSenderNode *theSender =
+                (NormSenderNode *)sender_tree.FindNodeById(nack.GetSenderId());
+            ASSERT(NULL != theSender);
+            fecM = theSender->GetFecFieldSize();
+            instId = theSender->GetInstanceId();
+            struct timeval grttResponse;
+            // When probe_tos is non-zero, GRTT feedback is in ACKs only
+            if (0 == probe_tos)
+            {
+                struct timeval currentTime;
+                ProtoSystemTime(currentTime);
+                theSender->CalculateGrttResponse(currentTime, grttResponse);
+            }
+            else
+            {
+                grttResponse.tv_sec = grttResponse.tv_usec = 0;
+            }
+            nack.SetGrttResponse(grttResponse);
+            break;
+        }
+        case NormMsg::ACK:
+        {
+            msg.SetSequence(0); // TBD - set per destination
+            isReceiverMsg = true;
+            NormAckMsg &ack = (NormAckMsg &)msg;
+            NormSenderNode *theSender;
+            if (IsServerListener())
+                theSender = client_tree.FindNodeByAddress(ack.GetDestination());
+            else
+                theSender = (NormSenderNode *)sender_tree.FindNodeById(ack.GetSenderId());
+            ASSERT(NULL != theSender);
+            fecM = theSender->GetFecFieldSize();
+            instId = theSender->GetInstanceId();
+            struct timeval grttResponse;
+            if ((0 == probe_tos) || (NormAck::CC == ack.GetAckType()))
+            {
+                struct timeval currentTime;
+                ProtoSystemTime(currentTime);
+                theSender->CalculateGrttResponse(currentTime, grttResponse);
+                ack.SetGrttResponse(grttResponse);
+                if (0 != probe_tos) sendRaw = true;
+            }
+            else
+            {
+                grttResponse.tv_sec = grttResponse.tv_usec = 0;
+            }
+            break;
+        }
         default:
             break;
-        }
-        msg.SetSequence(tx_sequence++); // (TBD) set for session dst msgs
-        break;
-    }
-    case NormMsg::NACK:
-    {
-        msg.SetSequence(0); // TBD - set per destination
-        isReceiverMsg = true;
-        NormNackMsg &nack = (NormNackMsg &)msg;
-        NormSenderNode *theSender =
-            (NormSenderNode *)sender_tree.FindNodeById(nack.GetSenderId());
-        ASSERT(NULL != theSender);
-        fecM = theSender->GetFecFieldSize();
-        instId = theSender->GetInstanceId();
-        struct timeval currentTime;
-        ProtoSystemTime(currentTime);
-        struct timeval grttResponse;
-        theSender->CalculateGrttResponse(currentTime, grttResponse);
-        nack.SetGrttResponse(grttResponse);
-        break;
-    }
-    case NormMsg::ACK:
-    {
-        msg.SetSequence(0); // TBD - set per destination
-        isReceiverMsg = true;
-        NormAckMsg &ack = (NormAckMsg &)msg;
-        NormSenderNode *theSender;
-        if (IsServerListener())
-            theSender = client_tree.FindNodeByAddress(ack.GetDestination());
-        else
-            theSender = (NormSenderNode *)sender_tree.FindNodeById(ack.GetSenderId());
-        ASSERT(NULL != theSender);
-        fecM = theSender->GetFecFieldSize();
-        instId = theSender->GetInstanceId();
-        struct timeval grttResponse;
-        struct timeval currentTime;
-        ProtoSystemTime(currentTime);
-        theSender->CalculateGrttResponse(currentTime, grttResponse);
-        ack.SetGrttResponse(grttResponse);
-        break;
-    }
-    default:
-        break;
     }
     // Fill in common message fields
     msg.SetSourceId(local_node_id);
@@ -4957,8 +5008,11 @@ NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg)
     else
     {
         unsigned int numBytes = msgSize;
-        bool result = tx_socket->SendTo(msg.GetBuffer(), numBytes, msg.GetDestination());
+        bool result;
-
+        if (sendRaw)
+            result = RawSendTo(msg.GetBuffer(), numBytes, msg.GetDestination(), probe_tos);
+        else
+            result = tx_socket->SendTo(msg.GetBuffer(), numBytes, msg.GetDestination());
         if (result)
         {
             if (numBytes == msgSize)
@@ -5031,6 +5085,73 @@ NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg)
     return MSG_SEND_OK;
 } // end NormSession::SendMessage()
 
+#ifdef ECN_SUPPORT
+bool NormSession::RawSendTo(const char* buffer, unsigned int& numBytes, const ProtoAddress& dstAddr, UINT8 trafficClass)
+{
+    // Send the message via proto_cap instead of UDP socket
+    // (Used for marking GRTT probing with different traffic class
+    //  to inform lower layer protocols to not delay the messages
+    //  via retransmission or other means).
+    UINT32 pcapBuffer[8192/4];  // TBD - Is this big enough???
+    UINT16* ethBuffer = (UINT16*)pcapBuffer + 2;  // offset for IP packet alignment
+    ProtoPktETH ethPkt(ethBuffer, 8192 - 2);
+    
+    // Ethernet source address will be set by ProtoCap::Forward() method
+    ProtoAddress etherDst;
+    if (dstAddr.IsMulticast())
+        etherDst.GetEthernetMulticastAddress(dstAddr);
+    else
+        etherDst = PROTO_ADDR_BROADCAST;
+    ethPkt.SetDstAddr(etherDst);
+    
+    if (ecn_enabled)
+    {
+        trafficClass |= ((UINT8)ProtoSocket::ECN_ECT0);  // set ECT0 bit
+        trafficClass &= ~((UINT8)ProtoSocket::ECN_ECT1); // clear ECT1 bit
+    }
+    
+    switch (dstAddr.GetType())
+    {
+        case ProtoAddress::IPv4:
+        {
+            ethPkt.SetType(ProtoPktETH::IP);
+            ProtoPktIPv4 ip4Pkt;
+            ip4Pkt.InitIntoBuffer(ethPkt.AccessPayload(), ethPkt.GetBufferLength() - ethPkt.GetHeaderLength());
+            ip4Pkt.SetTOS(trafficClass);
+            ip4Pkt.SetID((UINT16)rand());
+            ip4Pkt.SetTTL(ttl);
+            ip4Pkt.SetProtocol(ProtoPktIP::UDP);
+            ip4Pkt.SetSrcAddr(src_addr);
+            ip4Pkt.SetDstAddr(dstAddr);
+            ProtoPktUDP udpPkt(ip4Pkt.AccessPayload(), ip4Pkt.GetBufferLength() - ip4Pkt.GetHeaderLength());
+            udpPkt.SetSrcPort(GetTxPort());
+            udpPkt.SetDstPort(dstAddr.GetPort());
+            udpPkt.SetPayload(buffer, numBytes);
+            ip4Pkt.SetPayloadLength(udpPkt.GetLength());
+            udpPkt.FinalizeChecksum(ip4Pkt);
+            ethPkt.SetPayloadLength(ip4Pkt.GetLength());
+            break;
+        }
+        case ProtoAddress::IPv6:
+            ethPkt.SetType(ProtoPktETH::IPv6);
+            // IPv6 support TBD
+            return false;
+            break;
+        default:
+             PLOG(PL_ERROR, "NormSession::RawSendTo() error: invalid address type!\n");
+             return false;
+    } 
+    unsigned int ethBytes = ethPkt.GetLength();
+    bool result =  proto_cap->Forward((char*)ethPkt.AccessBuffer(), ethBytes);
+    if (!result)
+    {
+        PLOG(PL_WARN, "NormSession::RawSendTo() warning: proto_cap send failure!\n");
+        if (0 == ethBytes) numBytes = 0;
+    }
+    return result;
+}
+#endif  // NormSession::RawSendTo()
+
 void NormSession::SetGrttProbingInterval(double intervalMin, double intervalMax)
 {
     if ((intervalMin < 0.0) || (intervalMax < 0.0))