tcp-session.cc

Go to the documentation of this file.

/* -*-  Mode:C++; c-basic-offset:8; tab-width:8; indent-tabs-mode:t -*- */
/*
 * Copyright (c) 1997 Regents of the University of California.
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. All advertising materials mentioning features or use of this software
 *    must display the following acknowledgement:
 *      This product includes software developed by the Daedalus Research
 *      Group at the University of California Berkeley.
 * 4. Neither the name of the University nor of the Research Group may be
 *    used to endorse or promote products derived from this software without
 *    specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * $Header: /nfs/jade/vint/CVSROOT/ns-2/tcp/tcp-session.cc,v 1.18 2000/09/01 03:04:07 haoboy Exp $
 */

#include <stdlib.h>
#include <math.h>
#include "ip.h"
#include "flags.h"
#include "random.h"
#include "template.h"
#include "nilist.h"
#include "tcp.h"
#include "tcp-int.h"
#include "tcp-session.h"

/*
 * We separate TCP functionality into two parts: that having to do with 
 * providing a reliable, ordered byte-stream service, and that having to do with
 * congestion control and loss recovery. The former is done on a per-connection
 * basis and is implemented as part of IntTcpAgent ("integrated TCP"). The 
 * latter is done in an integrated fashion across multiple TCP connections, and
 * is implemented as part of TcpSessionAgent ("TCP session"). TcpSessionAgent is
 * derived from CorresHost ("correspondent host"), which keeps track of the 
 * state of all TCP (TCP/Int) connections to a host that it is corresponding 
 * with.
 *
 * The motivation for this separation of functionality is to make an ensemble of
 * connection more well-behaved than a set of independent TCP connections.
 * The packet loss rate is cut down and the chances of losses being recovered 
 * via data-driven techniques (rather than via timeouts) is improved. At the 
 * same time, we do not introduce any unnecessary coupling between the 
 * logically-independent byte-streams that the set of connections represents. 
 * This is in contrast to the coupling that is inherent in the multiplexing at 
 * the application layer of multiple byte-streams onto a single TCP connection.
 *
 * For questions/comments, please contact:
 *   Venkata N. Padmanabhan (padmanab@cs.berkeley.edu)
 *   http://www.cs.berkeley.edu/~padmanab
 */

static class TcpSessionClass : public TclClass {
public:
        TcpSessionClass() : TclClass("Agent/TCP/Session") {}
        TclObject* create(int, const char*const*) {
                return (new TcpSessionAgent()); 
        }
} class_tcpsession;

TcpSessionAgent::TcpSessionAgent() : CorresHost(), 
        rtx_timer_(this), burstsnd_timer_(this), sessionSeqno_(0),
        last_send_time_(-1), curConn_(0), numConsecSegs_(0), 
        schedDisp_(FINE_ROUND_ROBIN), wtSum_(0), dynWtSum_(0) 
{
        bind("ownd_", &ownd_);
        bind("owndCorr_", &owndCorrection_);
        bind_bool("proxyopt_", &proxyopt_);
        bind_bool("fixedIw_", &fixedIw_);
        bind("schedDisp_", &schedDisp_);
        bind_bool("disableIntLossRecov_", &disableIntLossRecov_);

        sessionList_.append(this);
}

int
TcpSessionAgent::command(int argc, const char*const* argv)
{
        if (argc == 2) {
                if (!strcmp(argv[1], "resetwt")) {
                        Islist_iter<IntTcpAgent> conn_iter(conns_);
                        IntTcpAgent *tcp;

                        while ((tcp = conn_iter()) != NULL) 
                                tcp->wt_ = 1;
                        wtSum_ = conn_iter.count();
                        return (TCL_OK);
                }
        }
        return (CorresHost::command(argc, argv));
}

void
SessionRtxTimer::expire(Event*)
{
        a_->timeout(TCP_TIMER_RTX);
}

void
SessionResetTimer::expire(Event*)
{
        a_->timeout(TCP_TIMER_RESET);
}

void
SessionBurstSndTimer::expire(Event*)
{
        a_->timeout(TCP_TIMER_BURSTSND);
}

void
TcpSessionAgent::reset_rtx_timer(int /*mild*/, int backoff)
{
        if (backoff)
                rtt_backoff();
        set_rtx_timer();
        rtt_active_ = 0;
}

void
TcpSessionAgent::set_rtx_timer()
{
        if (rtx_timer_.status() == TIMER_PENDING)
                rtx_timer_.cancel();
        if (reset_timer_.status() == TIMER_PENDING)
                reset_timer_.cancel();
        if (fs_enable_ && fs_mode_)
                reset_timer_.resched(rtt_exact_timeout());
        else
                rtx_timer_.resched(rtt_timeout());
}

void
TcpSessionAgent::cancel_rtx_timer()
{
        rtx_timer_.force_cancel();
        reset_timer_.force_cancel();
}

void
TcpSessionAgent::cancel_timers()
{
        rtx_timer_.force_cancel();
        reset_timer_.force_cancel();
        burstsnd_timer_.force_cancel();
        delsnd_timer_.force_cancel();
}

int
TcpSessionAgent::fs_pkt() {
        return (fs_enable_ && fs_mode_ && sessionSeqno_-1 >= fs_startseq_ &&
                sessionSeqno_-1 < fs_endseq_);
}

void
TcpSessionAgent::rtt_update_exact(double tao)
{
        double g = 1/8; /* gain used for smoothing rtt */
        double h = 1/4; /* gain used for smoothing rttvar */
        double delta;

        if (t_exact_srtt_ != 0) {
                delta = tao - t_exact_srtt_;
                if (delta < 0)
                        delta = -delta;
                /* update the fine-grained estimate of the smoothed RTT */
                if (t_exact_srtt_ != 0) 
                        t_exact_srtt_ = g*tao + (1-g)*t_exact_srtt_;
                else
                        t_exact_srtt_ = tao;
                /* update the fine-grained estimate of mean deviation in RTT */
                delta -= t_exact_rttvar_;
                t_exact_rttvar_ += h*delta;
        }
        else {
                t_exact_srtt_ = tao;
                t_exact_rttvar_ = tao/2;
        }
}       

void
TcpSessionAgent::newack(Packet *pkt) 
{
        double now = Scheduler::instance().clock();
        Islist_iter<Segment> seg_iter(seglist_);
        hdr_tcp *tcph = hdr_tcp::access(pkt);
        hdr_flags *fh = hdr_flags::access(pkt);

        if (!fh->no_ts_) {
                /* if the timestamp option is being used */
                if (ts_option_) {
                        rtt_update(now - tcph->ts_echo());
                        rtt_update_exact(now - tcph->ts_echo());
                }
                /* if segment being timed just got acked */
                if (rtt_active_ && rtt_seg_ == NULL) {
                        t_backoff_ = 1;
                        rtt_active_ = 0;
                        if (!ts_option_)
                                rtt_update(now - rtt_ts_);
                }
        }
        if (seg_iter.count() > 0)
                set_rtx_timer();
        else
                cancel_rtx_timer();
}

void
TcpSessionAgent::timeout(int tno)
{
        if (tno == TCP_TIMER_BURSTSND)
                send_much(NULL,0,0);
        else if (tno == TCP_TIMER_RESET) {
                Islist_iter<Segment> seg_iter(seglist_);
                Segment *curseg;
                Islist_iter<IntTcpAgent> conn_iter(conns_);
                IntTcpAgent *curconn;

                fs_mode_ = 0;
                if (seg_iter.count() == 0 && !slow_start_restart_) {
                        return;
                }
                recover_ = sessionSeqno_ - 1;
                last_cwnd_action_ = CWND_ACTION_TIMEOUT;
                ownd_ = 0;
                owndCorrection_ = 0;
                while ((curconn = conn_iter()) != NULL) {
                        curconn->maxseq_ = curconn->highest_ack_;
                        curconn->t_seqno_ = curconn->highest_ack_ + 1;
                        curconn->recover_ = curconn->maxseq_;
                        curconn->last_cwnd_action_ = CWND_ACTION_TIMEOUT;
                }
                while ((curseg = seg_iter()) != NULL) {
                        /* XXX exclude packets sent "recently"? */
                        curseg->size_ = 0;
                }

                /* 
                 * If first pkt sent before fast start has not gotten through, 
                 * treat this as a regular rtx timeout. Otherwise, close cwnd
                 * and reset timer but don't back off timer.
                 */
                if (connWithPktBeforeFS_) {
                        connWithPktBeforeFS_ = NULL;
                        timeout(TCP_TIMER_RTX);
                }
                else {
                        slowdown(CLOSE_CWND_INIT);
                        reset_rtx_timer(0,0);
                        send_much(NULL, 0, TCP_REASON_TIMEOUT);
                }
        }
        else if (tno == TCP_TIMER_RTX) {
                Islist_iter<Segment> seg_iter(seglist_);
                Segment *curseg;
                Islist_iter<IntTcpAgent> conn_iter(conns_);
                IntTcpAgent *curconn;

                if (seg_iter.count() == 0 && !slow_start_restart_) {
                        return;
                }
                recover_ = sessionSeqno_ - 1;
                last_cwnd_action_ = CWND_ACTION_TIMEOUT;
                if (seg_iter.count() == 0 && restart_bugfix_) {
                        slowdown(CLOSE_CWND_INIT);
                        reset_rtx_timer(0,0);
                }
                else {
                        slowdown(CLOSE_CWND_RESTART|CLOSE_SSTHRESH_HALF);
                        reset_rtx_timer(0,1);
                }
                nrexmit_++;
                ownd_ = 0;
                owndCorrection_ = 0;
                while ((curconn = conn_iter()) != NULL) {
                        curconn->t_seqno_ = curconn->highest_ack_ + 1;
                        curconn->recover_ = curconn->maxseq_;
                        curconn->last_cwnd_action_ = CWND_ACTION_TIMEOUT;
                }
                while ((curseg = seg_iter()) != NULL) {
                        /* XXX exclude packets sent "recently"? */
                        curseg->size_ = 0;
                }

                send_much(NULL, 0, TCP_REASON_TIMEOUT);
        }
        else
                printf("TcpSessionAgent::timeout(): ignoring unknown timer %d\n", tno);
}

Segment* 
TcpSessionAgent::add_pkts(int size, int seqno, int sessionSeqno, int daddr, 
                          int dport, int sport, double ts, IntTcpAgent *sender)
{
        /*
         * set rtx timer afresh either if it is not set now or if there are no 
         * data packets outstanding at this time
         */
        if (!(rtx_timer_.status() == TIMER_PENDING) || seglist_.count() == 0)
                set_rtx_timer();
        last_seg_sent_ = CorresHost::add_pkts(size, seqno, sessionSeqno, daddr, dport, sport, ts, sender);
        return last_seg_sent_;
}
                
void
TcpSessionAgent::add_agent(IntTcpAgent *agent, int size, double winMult, 
                           int winInc, int ssthresh)
{
        CorresHost::add_agent(agent,size,winMult,winInc,ssthresh);
        wtSum_ += agent->wt_;
        reset_dyn_weights();
}

int
TcpSessionAgent::window()
{
        if (maxcwnd_ == 0)
                return (int(cwnd_));
        else
                return (int(min(cwnd_,maxcwnd_)));
}

void
TcpSessionAgent::set_weight(IntTcpAgent *tcp, int wt)
{
        wtSum_ -= tcp->wt_;
        tcp->wt_ = wt;
        wtSum_ += tcp->wt_;
}
                        
void
TcpSessionAgent::reset_dyn_weights()
{
        IntTcpAgent *tcp;
        Islist_iter<IntTcpAgent> conn_iter(conns_);

        while ((tcp = conn_iter()) != NULL)
                tcp->dynWt_ = tcp->wt_;
        dynWtSum_ = wtSum_;
}

IntTcpAgent *
TcpSessionAgent::who_to_snd(int how)
{
        int i = 0;
        switch (how) {
        /* fine-grained interleaving of connections (per pkt) */
        case FINE_ROUND_ROBIN: { 
                IntTcpAgent *next;
                int wtOK = 0;

                if (dynWtSum_ == 0) 
                        reset_dyn_weights();
                do {
                        wtOK = 0;
                        if ((next = (*connIter_)()) == NULL) {
                                connIter_->set_cur(connIter_->get_last());
                                next = (*connIter_)();
                        }
                        i++;
                        if (next && next->dynWt_>0) {
                                next->dynWt_--;
                                dynWtSum_--;
                                wtOK = 1;
                        }
                } while (next && (!next->data_left_to_send() || !wtOK)
                         && (i < connIter_->count()));
                if (!next->data_left_to_send())
                        next = NULL;
                return next;
        }
        /* coarse-grained interleaving across connections (per block of pkts) */
        case COARSE_ROUND_ROBIN: {
                int maxConsecSegs;
                if (curConn_)
                        maxConsecSegs = (window()*curConn_->wt_)/wtSum_;
                if (curConn_ && numConsecSegs_++ < maxConsecSegs && 
                        curConn_->data_left_to_send())
                        return curConn_;
                else {
                        numConsecSegs_ = 0;
                        curConn_ = who_to_snd(FINE_ROUND_ROBIN);
                        if (curConn_)
                                numConsecSegs_++;
                }
                return curConn_;
        }
        case RANDOM: {
                IntTcpAgent *next;
                
                do {
                        int foo = int(Random::uniform() * nActive_ + 1);
                        
                        connIter_->set_cur(connIter_->get_last());
                        
                        for (;foo > 0; foo--)
                                (*connIter_)();
                        next = (*connIter_)();
                } while (next && !next->data_left_to_send());
                return(next);
        }
        default:
                return NULL;
        }
}

void
TcpSessionAgent::send_much(IntTcpAgent* /*agent*/, int force, int reason) 
{
        int npackets = 0;
        Islist_iter<Segment> seg_iter(seglist_);

        if (reason != TCP_REASON_TIMEOUT &&
            burstsnd_timer_.status() == TIMER_PENDING)
                return;
        /* no outstanding data and idle time >= t_rtxcur_ */
        if ((seg_iter.count() == 0) && (last_send_time_ != -1) &&
            (Scheduler::instance().clock() - last_send_time_ >= t_rtxcur_)) {
                if (slow_start_restart_ && restart_bugfix_)
                        slowdown(CLOSE_CWND_INIT);
                else if (slow_start_restart_)
                        slowdown(CLOSE_CWND_RESTART|CLOSE_SSTHRESH_HALF);
                else if (fs_enable_) {
                        if (cwnd_ < ssthresh_)
                                cwnd_ = int(cwnd_/2);
                        else
                                cwnd_ -= 1;
                        fs_startseq_ = sessionSeqno_ + 1;
                        fs_endseq_ = sessionSeqno_ + window();
                        fs_mode_ = 1;
                }
        }

        while (ok_to_snd(size_)) {
                {
                        IntTcpAgent *sender = who_to_snd(schedDisp_);
                        if (sender) {
                                /*
                                 * remember the connection over which the first
                                 * packet just before fast start is sent
                                 */
                                if (fs_enable_ && fs_mode_ && 
                                    sessionSeqno_ == fs_startseq_)
                                        connWithPktBeforeFS_ = sender;
                                /* if retransmission */
                                /* XXX we pick random conn even if rtx timeout */
                                if (sender->t_seqno_ < sender->maxseq_) {
                                        int i = 
                                findSessionSeqno(sender, sender->t_seqno_);
                                        removeSessionSeqno(i);
                                        sender->send_one(i);
                                }
                                else {
                                        sender->send_one(sessionSeqno_++);
                                        if (!rtt_active_) {
                                                rtt_active_ = 1;
                                                rtt_seg_ = last_seg_sent_;
                                        }
                                }
                                npackets++;
                        }
                        else
                                break;
                }
                reason = 0;
                force = 0;
                if (maxburst_ && npackets == maxburst_) {
                        if (ok_to_snd(size_))
                                burstsnd_timer_.resched(t_exact_srtt_*maxburst_/window());
                        break;
                }
        }
        if (npackets > 0)
                last_send_time_ = Scheduler::instance().clock();
}

void
TcpSessionAgent::recv(IntTcpAgent *agent, Packet *pkt, int amt_data_acked)
{
        hdr_tcp *tcph = hdr_tcp::access(pkt);

        if (hdr_flags::access(pkt)->ecnecho() && ecn_)
                quench(1, agent, tcph->seqno());
        clean_segs(size_, pkt, agent, sessionSeqno_,amt_data_acked);
        /* XXX okay to do this after clean_segs? */
        /* if new data acked and this is not a partial ack */
        if (amt_data_acked > 0 && (tcph->seqno() >= agent->recover_ ||
           agent->last_cwnd_action_ != CWND_ACTION_DUPACK /* XXX 1*/) 
            && !dontIncrCwnd_) {
                int i = count_bytes_acked_ ? amt_data_acked:1;
                while (i-- > 0)
                        opencwnd(size_,agent);
        }
        dontIncrCwnd_ = 0;
        if (amt_data_acked > 0) {
                if (fs_enable_ && fs_mode_ && connWithPktBeforeFS_ == agent)
                        connWithPktBeforeFS_ = NULL;
                newack(pkt);
        }
        Packet::free(pkt);
        send_much(NULL,0,0);
}
        
void
TcpSessionAgent::setflags(Packet *pkt)
{
        hdr_flags *hf = hdr_flags::access(pkt);
        if (ecn_)
                hf->ect() = 1;
}

int
TcpSessionAgent::findSessionSeqno(IntTcpAgent *sender, int seqno)
{
        Islist_iter<Segment> seg_iter(seglist_);
        Segment *cur;
        int min = sessionSeqno_;
        
        while ((cur = seg_iter()) != NULL) {
                if (sender == cur->sender_ && cur->seqno_ >= seqno && 
                    cur->sessionSeqno_ < min)
                        min = cur->sessionSeqno_;
        }
        if (min == sessionSeqno_) {
                printf("In TcpSessionAgent::findSessionSeqno: search unsuccessful\n");
                min = sessionSeqno_ - 1;
        }
        return (min);
}


void
TcpSessionAgent::removeSessionSeqno(int sessionSeqno) 
{
        Islist_iter<Segment> seg_iter(seglist_);
        Segment *cur, *prev=NULL;
        
        while ((cur = seg_iter()) != NULL) {
                if (cur->sessionSeqno_ == sessionSeqno) {
                        seglist_.remove(cur, prev);
                        adjust_ownd(cur->size_);
                        return;
                }
                prev = cur;
        }
        printf("In removeSessionSeqno(): unable to find segment with sessionSeqno = %d\n", sessionSeqno);
}

void
TcpSessionAgent::quench(int how, IntTcpAgent *sender, int seqno)
{
        int i = findSessionSeqno(sender,seqno);

        if (i > recover_) {
                recover_ = sessionSeqno_ - 1;
                last_cwnd_action_ = CWND_ACTION_ECN;
                sender->recover_ = sender->maxseq_;
                sender->last_cwnd_action_ = CWND_ACTION_ECN;
                closecwnd(how,sender);
        }
}

void
TcpSessionAgent::traceVar(TracedVar* v)
{
        double curtime;
        Scheduler& s = Scheduler::instance();
        char wrk[500];
        int n;
        
        curtime = &s ? s.clock() : 0;
        if (!strcmp(v->name(), "ownd_") || !strcmp(v->name(), "owndCorr_")) {
                if (!strcmp(v->name(), "ownd_"))
                        sprintf(wrk,"%-8.5f %-2d %-2d %-2d %-2d %s %-6.3f", 
                                curtime, addr(), port(), daddr(), dport(),
                                v->name(), double(*((TracedDouble*) v))); 
                else if (!strcmp(v->name(), "owndCorr_"))
                        sprintf(wrk,"%-8.5f %-2d %-2d %-2d %-2d %s %d", 
                                curtime, addr(), port(), daddr(), dport(),
                                v->name(), int(*((TracedInt*) v))); 
                n = strlen(wrk);
                wrk[n] = '\n';
                wrk[n+1] = 0;
                if (channel_)
                        (void)Tcl_Write(channel_, wrk, n+1);
                wrk[n] = 0;
        }
        else
                TcpAgent::traceVar(v);
}

                        

---