From c738eea1a9a89b23ffb4f57d27df0799b8068bf2 Mon Sep 17 00:00:00 2001 From: bebopagogo Date: Sat, 6 Jun 2020 12:41:58 -0400 Subject: [PATCH] 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 | 32 ++- examples/normStreamer.cpp | 30 ++- include/normApi.h | 4 + include/normMessage.h | 3 +- include/normObject.h | 5 +- include/normSession.h | 19 +- src/common/normApi.cpp | 13 ++ src/common/normFile.cpp | 2 +- src/common/normMessage.cpp | 12 ++ src/common/normObject.cpp | 12 +- src/common/normSession.cpp | 393 ++++++++++++++++++++++++------------- 11 files changed, 374 insertions(+), 151 deletions(-) diff --git a/examples/normCast.cpp b/examples/normCast.cpp index 9a61d58..95aa3fc 100644 --- a/examples/normCast.cpp +++ b/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 {send &| recv }\n" " [addr [/]][interface ][ack [,,...]\n" - " [cc|cce|ccl|rate ]\n" + " [cc|cce|ccl|rate ][ptos ]\n" " [debug ][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) diff --git a/examples/normStreamer.cpp b/examples/normStreamer.cpp index 791c4cd..b7db6ad 100644 --- a/examples/normStreamer.cpp +++ b/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 -- 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 -- 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 -- 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; diff --git a/include/normApi.h b/include/normApi.h index cc6e2bc..281f80e 100644 --- a/include/normApi.h +++ b/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); diff --git a/include/normMessage.h b/include/normMessage.h index ae68765..7abe05a 100644 --- a/include/normMessage.h +++ b/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: diff --git a/include/normObject.h b/include/normObject.h index 413ec20..2446e79 100644 --- a/include/normObject.h +++ b/include/normObject.h @@ -52,10 +52,7 @@ class NormObject // This must be reset after each update void SetNotifyOnUpdate(bool state) - { - TRACE("enter SetNotifyOnUpdate(%d) ...\n", state); - notify_on_update = state; - } + {notify_on_update = state;} // Object information NormObject::Type GetType() const {return type;} diff --git a/include/normSession.h b/include/normSession.h index e62cab5..38971c6 100644 --- a/include/normSession.h +++ b/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 diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index b076822..a1a592d 100644 --- a/src/common/normApi.cpp +++ b/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) diff --git a/src/common/normFile.cpp b/src/common/normFile.cpp index efa7124..dce2d7a 100644 --- a/src/common/normFile.cpp +++ b/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 diff --git a/src/common/normMessage.cpp b/src/common/normMessage.cpp index 900ecaf..486e93b 100644 --- a/src/common/normMessage.cpp +++ b/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, diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index eb37f06..a1bf6d4 100644 --- a/src/common/normObject.cpp +++ b/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, diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index c61ed1c..e75f797 100644 --- a/src/common/normSession.cpp +++ b/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), - grtt_interval_min(DEFAULT_GRTT_INTERVAL_MIN), + probe_tos(0), grtt_interval(0.5), 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(); - return false; - } - rx_cap->SetListener(this, &NormSession::OnPktCapture); - rx_cap->SetNotifier(session_mgr.GetChannelNotifier()); - if (!rx_cap->Open(('\0' != interface_name[0]) ? interface_name : NULL)) - { - PLOG(PL_FATAL, "NormSession::Open() error: unable to open ProtoCap device '%s'!\n", (('\0' != interface_name[0]) ? interface_name : "(null)")); - Close(); - 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; - } + PLOG(PL_FATAL, "NormSession::Open() error: unable to create ProtoCap device!\n"); + Close(); + return false; + } + if (ecn_enabled) + { + rx_socket.StopInputNotification(); // Disable rx_socket (keep open so mcast JOIN holds) + proto_cap->StartInputNotification(); + } + else + { + proto_cap->StopInputNotification(); } - 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) - { - rx_cap->Close(); - delete rx_cap; - rx_cap = NULL; - } + CloseProtoCap(); #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) - { - rx_cap->Close(); - delete rx_cap; - rx_cap = NULL; - } + if (NULL != proto_cap) + proto_cap->StopInputNotification(); #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 ¤tTime, @@ -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::DATA: - { - NormObjectMsg &objMsg = static_cast(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(msg); - ((NormCmdMsg &)msg).SetInstanceId(instId); - switch (cmd.GetFlavor()) + case NormMsg::INFO: + case NormMsg::DATA: { - case NormCmdMsg::CC: - { - NormCmdCCMsg &ccMsg = static_cast(cmd); - struct timeval currentTime; - ProtoSystemTime(currentTime); - ccMsg.SetSendTime(currentTime); - isProbe = true; + NormObjectMsg &objMsg = static_cast(msg); + objMsg.SetInstanceId(instId); + msg.SetSequence(tx_sequence++); // (TBD) set for session dst msgs if (syn_status) - ccMsg.SetSyn(); + objMsg.SetFlag(NormObjectMsg::FLAG_SYN); break; } - case NormCmdMsg::SQUELCH: + case NormMsg::CMD: + { + NormCmdMsg &cmd = static_cast(msg); + ((NormCmdMsg &)msg).SetInstanceId(instId); + switch (cmd.GetFlavor()) + { + case NormCmdMsg::CC: + { + NormCmdCCMsg &ccMsg = static_cast(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))