scheduler.c

Go to the documentation of this file.

/* Packet transmission scheduler.
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <time.h>
#include <semaphore.h>
#include <pthread.h>
#include <sys/socket.h>
#include <linux/tcp.h>
#include <linux/ip.h>
#include <linux/in.h>

#include "tcpproxy.h"
#include "misc.h"
#include "tcptimer.h"
#include "tcp.h"
#include "scheduler.h"

enum queue_state
{
    QUEUE_STATE_FREE = 0,
    QUEUE_STATE_USED,
    QUEUE_STATE_CLOSED
} Tqueue_state;

/* Sent packet record. */
typedef struct packet_sent {
    struct packet_queue *q;          /* Queue packet was sent from. */
    unsigned seq;                    /* Sequence number of packet.  */
    unsigned size;                   /* Size of packet.             */
    long mtime;                      /* Time sent in miliseconds.   */
    struct packet_sent *prev;        /* Previous record.            */
    struct packet_sent *next;        /* Next record.                */
} Tpacket_sent;

/* Circular FIFO packet queue. */
typedef struct packet_queue {
    enum queue_state state; /* State of the queue.                           */
    Tpacket_data **tab;     /* Array of pointers to packet data blocks.      */
    int max;                /* Max size of queue.                            */
    int head;               /* Head position.                                */
    int tail;               /* Tail position.                                */
    unsigned in_flight;     /* Number of data bytes in flight over the link. */
    unsigned dc;            /* Deficit counter for DRR scheduling.           */
    unsigned weight;        /* Weight for DRR scheduling.                    */
    Tpacket_sent *sent_hd;  /* Head of linked list of sent packets.          */
} Tpacket_queue;

static int sock;                              /* Socket for tx packets.      */
static Tpacket_queue tcp_pq[TCP_MAX_QUEUES];  /* Array of TCP packet queues. */
static Tpacket_queue hipri_pq;                /* Queue for hi pri. packets.  */
static Tpacket_queue lowpri_pq;               /* Queue for low pri. packets. */
static pthread_mutex_t sched_mut;             /* Scheduler mutex.            */
static sem_t send_sem;                        /* Send packet semaphore.      */
static long queued_pkts;                      /* Total queued packets.       */
static unsigned tcp_in_flight;       /* No. TCP bytes in flight on the link. */
static int max_tcp_in_flight = TCP_WIN_CLAMP_SIZE;  /* Max data on the link. */

/*----------------------------------------------------------------------
 * alloc_queues -  perform scheduler initialisation.
 *----------------------------------------------------------------------
 */
void init_sched()
{
    int i;
    Tpacket_queue *q;
    int old_flags;
    int on = 1; 

    //DPRINT("init_sched: entered.\n");

    sock = socket(AF_INET, SOCK_RAW, IPPROTO_RAW);

    if(sock == -1) {
        EPRINT("init_sched: error opening raw datagram socket.\n");
        perror("init_sched");
        safe_exit(EXIT_ERR_SOCKET);
    }

    if(setsockopt(sock, IPPROTO_IP, IP_HDRINCL, &on, sizeof(on))) {
        EPRINT("init_sched: error setting IP_HDRINCL.\n");
        perror("init_sched");
        safe_exit(EXIT_ERR_SOCKET);
    }

    old_flags = fcntl (sock, F_GETFL, 0);  /* Get socket flags. */
    if (old_flags == -1) {
        EPRINT("init_sched: error reading socket flags.\n");
        perror("init_sched");
        safe_exit(EXIT_ERR_SOCKET);
    }
    old_flags &= ~O_NONBLOCK;  /* Clear non-blocking flag. */
    if(fcntl(sock, F_SETFL, old_flags) == -1)  /* Set socket flags. */
    {
        EPRINT("init_sched: error setting socket flags.\n");
        perror("init_sched");
        safe_exit(EXIT_ERR_SOCKET);     
    }

    pthread_mutex_init(&sched_mut, NULL);
    tcp_in_flight = 0;

    if(sem_init(&send_sem, 0, 0)) {
        EPRINT("init_sched: error initialising semaphore.\n");
        safe_exit(EXIT_ERR_SEM_INIT);
    }

    /* Initialise ICMP and other packet queues. */
    hipri_pq.state = QUEUE_STATE_USED;
    hipri_pq.tab = (Tpacket_data **) xmalloc(sizeof(Tpacket_data*)*QUEUE_SIZE_HI_PRI);
    hipri_pq.max = QUEUE_SIZE_HI_PRI;
    hipri_pq.head = 0;
    hipri_pq.tail = 0;
    lowpri_pq.state = QUEUE_STATE_USED;
    lowpri_pq.tab = (Tpacket_data **) xmalloc(sizeof(Tpacket_data*)*QUEUE_SIZE_LOW_PRI);
    lowpri_pq.max = QUEUE_SIZE_LOW_PRI;
    lowpri_pq.head = 0;
    lowpri_pq.tail = 0;
        
    /* Initialise the tcp packet queues. */
    for(i=0; i<TCP_MAX_QUEUES; i++) {
        q = &tcp_pq[i];
        q->state = QUEUE_STATE_FREE;
        q->tab = (Tpacket_data **) xmalloc(sizeof(Tpacket_data*)*QUEUE_SIZE_TCP);
        q->max = QUEUE_SIZE_TCP;
        q->head = 0;
        q->tail = 0;
        q->in_flight = 0;
        q->dc = 0;
        q->weight = QUEUE_WEIGHT_NORMAL;
        q->sent_hd = NULL;
    }

    queued_pkts = 0;
}

/*----------------------------------------------------------------------
 * queue_size -  return size of selected queue.
 *----------------------------------------------------------------------
 */
inline int queue_size(Tpacket_queue *q)
{
    return ((q->tail >= q->head) ?
            (q->tail - q->head) :
            (q->tail + q->max - q->head));
}

/*----------------------------------------------------------------------
 * queue_empty -  return true if the queue is empty.
 *----------------------------------------------------------------------
 */
inline int queue_empty(Tpacket_queue *q)
{
    return (q->head == q->tail);
}

/*----------------------------------------------------------------------
 * packet_timeout -  called by tcp_timer when a packet times out;
 *                   the packet may have already been acknowledged, in
 *                   which case the record of it is merely deallocated.
 *----------------------------------------------------------------------
 */
void packet_timeout(void *arg)
{
    Tpacket_sent *sent = (Tpacket_sent*)arg;
    assert(sent != NULL);

    mutex_lock("scheduler", &sched_mut);

    if(sent->size > 0) {  /* Sent record still active. */
        Tpacket_queue *q = sent->q;
        DPRINT("packet_timeout: sent record still active, seq=%u, size=%u.\n", sent->seq, sent->size);
        if(q->sent_hd == sent) {
            q->sent_hd = sent->next;
        } else {
            assert(sent->prev != NULL);
            sent->prev->next = sent->next;
        }
        assert(sent->size <= q->in_flight);
        q->in_flight -= sent->size;
        tcp_in_flight -= sent->size;
    } else {
        //DPRINT("packet_timeout: sent record inactive.\n");
    }
    xfree(sent);
    mutex_unlock("scheduler", &sched_mut);    
}

/*----------------------------------------------------------------------
 * add_tcp_sent -  add a record of a tcp packet to the queue's sent list;
 *                 assumes sched_mut HELD.
 *----------------------------------------------------------------------
 */
void add_tcp_sent(Tpacket_data *pd, unsigned size, Tpacket_queue *q)
{
    Tpacket_sent *sent;
    Tpacket_sent *ptr, *prev_ptr;

    //DPRINT("add_tcp_sent: entered, size=%u.\n", size);
    assert(size > 0);    

    sent = (Tpacket_sent*) xmalloc(sizeof(Tpacket_sent));
    sent->q = q;
    sent->seq = ntohl(pd->trans.tcp->seq);
    sent->size = size;
    sent->mtime = get_mclock();
    sent->prev = sent->next = NULL;
    
    prev_ptr = NULL;
    for(ptr=q->sent_hd; ptr; ptr=ptr->next) {
        if(seq_cmp(sent->seq, ptr->seq) < 0)  /* Need to insert event before ptr. */
            break;
        prev_ptr = ptr;
    }

    if(prev_ptr) {
        prev_ptr->next = sent;
        sent->prev = prev_ptr;
    } else {
        q->sent_hd = sent;
    }
    sent->next = ptr;
    if(ptr)
        ptr->prev = sent;

    add_timer(TCP_PACKET_TIMEOUT, packet_timeout, sent);
}

/*----------------------------------------------------------------------
 * alloc_tcp_queue -  allocate a tcp output queue;
 *                    returns the number of the allocated queue, or -1
 *                    if there are no free queues.
 *----------------------------------------------------------------------
 */
int alloc_tcp_queue(unsigned weight)
{
    int i;
    mutex_lock("alloc_tcp_queue", &sched_mut);

    for(i=0; i<TCP_MAX_QUEUES; i++) {
        if(tcp_pq[i].state == QUEUE_STATE_FREE)
            break;
    }
    if(i < TCP_MAX_QUEUES) {
        tcp_pq[i].state = QUEUE_STATE_USED;
        tcp_pq[i].weight = weight;
        DPRINT("alloc_tcp_queue: allocated tcp queue %d with weight %u.\n", i, weight);
    } else {
        DPRINT("alloc_tcp_queue: no free queues.\n");
        i = -1;
    }
    mutex_unlock("alloc_tcp_queue", &sched_mut);

    return i;
}

/*----------------------------------------------------------------------
 * close_tcp_queue -  puts tcp queue into the closed state, so that it
 *                    will be deallocated when it becomes empty and has
 *                    no data in flight.
 *----------------------------------------------------------------------
 */
int close_tcp_queue(int queue_num)
{
    DPRINT("close_tcp_queue: queue %d\n", queue_num);
    assert(queue_num >= 0 && queue_num < TCP_MAX_QUEUES);

    mutex_lock("close_tcp_queue", &sched_mut);

    assert(tcp_pq[queue_num].state == QUEUE_STATE_USED);
    tcp_pq[queue_num].state = QUEUE_STATE_CLOSED;

    mutex_unlock("close_tcp_queue", &sched_mut);

    return 0;
}

/*----------------------------------------------------------------------
 * clear_tcp_queue -  clears a tcp queue, discarding any packets
 *                    still in the queue;
 *                    returns the number of packets discarded;
 *                    assumes sched_mut HELD.
 *----------------------------------------------------------------------
 */
int clear_tcp_queue(int queue_num)
{
    Tpacket_queue *q;
    Tpacket_sent *sent;
    int queued;
    int sent_rec;
    unsigned unacked_bytes;
    int i;

    assert(queue_num >= 0 && queue_num < TCP_MAX_QUEUES);

    q = &tcp_pq[queue_num];
    queued = queue_size(q);
    assert(queued >= 0);

    DPRINT("clear_tcp_queue: clearing queue %d, discarding %d packets.\n",
           queue_num, queued);

    unacked_bytes = 0;
    sent_rec = 0;
    while(q->sent_hd != NULL) {
        sent = q->sent_hd;
        //DPRINT("clear_tcp_queue: sent packet record, seq=%u, size=%u.\n",
        //       sent->seq, sent->size);
        unacked_bytes += sent->size;
        q->sent_hd = sent->next;
        sent_rec++;
        sent->size = 0;
        /* If the timer is successfully cleared (i.e. it is not currently being
         * activated), then free sent. Otherwise, sent will be freed by packet_timeout. */
        if(clear_timer(packet_timeout, sent) != -1) {
            xfree(sent);
            //DPRINT("clear_tcp_queue: cleared timer.\n");
        }
    }
    if(DEBUG && sent_rec > 0)
        DPRINT("clear_tcp_queue: cleared %d sent records.\n", sent_rec);

    if(q->in_flight > 0)
    {
        DPRINT("clear_tcp_queue: in_flight=%u, unacked_bytes=%u, ", q->in_flight, unacked_bytes);
        if(q->in_flight != unacked_bytes) {
            DPRINT("values do NOT match.\n");
        } else {
            DPRINT("values match ok.\n");
        }
        tcp_in_flight -= q->in_flight;  /* Decrement global in flight counter. */
        q->in_flight = 0;
    }

    for(i=0; i<queued; i++) { /* Throw away all packets. */
        dealloc_packet_data(q->tab[q->head]);
        q->head = (q->head+1) % q->max;
        queued_pkts--;
    }
    assert(queue_empty(q));
    assert(queued_pkts >= 0);

    sem_post(&send_sem);  /* Signal send semaphore, as we may have reduced tcp_in_flight. */

    return queued;
}

/*----------------------------------------------------------------------
 * reset_tcp_queue -  clears all packets from a TCP queue - called when
 *                    an RST is seen.
 *-----------------------------------------------------------------------
 */
int reset_tcp_queue(int queue_num)
{
    //DPRINT("reset_tcp_queue: queue: %d\n", queue_num);
    assert(queue_num >= 0 && queue_num < TCP_MAX_QUEUES);

    mutex_lock("reset_tcp_queue", &sched_mut);

    assert(tcp_pq[queue_num].state == QUEUE_STATE_USED);
    clear_tcp_queue(queue_num);

    mutex_unlock("reset_tcp_queue", &sched_mut);

    return 0;
}

/*----------------------------------------------------------------------
 * check_dealloc_tcp_queue -  check if a tcp queue is ready to be
 *                            deallocated, and calls clear_tcp_queue.
 *                            returns 1 if the queue is cleared, zero
 *                            otherwise.
 *                            assumes sched_mut HELD.
 *----------------------------------------------------------------------
 */
inline int check_dealloc_tcp_queue(int queue_num)
{
    Tpacket_queue *q;
    //DPRINT("check_dealloc_tcp_queue: queue_num=%d.\n", queue_num);
    assert(queue_num >= 0 && queue_num < TCP_MAX_QUEUES);
    q = &tcp_pq[queue_num];
    if(q->state == QUEUE_STATE_CLOSED && queue_empty(q) && q->in_flight == 0) {
        DPRINT("check_dealloc_tcp_queue: deallocating queue %d\n", queue_num);
        assert(q->head == q->tail);
        q->state = QUEUE_STATE_FREE;
        return 1;
    } else {
        return 0;
    }
}

/*----------------------------------------------------------------------
 * cleanup_sched -  destroy objects and deallocate memory.
 *----------------------------------------------------------------------
 */
void cleanup_sched()
{
    Tpacket_queue *q;
    int i;

    //DPRINT("cleanup_sched: entered.\n");

    for(i=0; i<TCP_MAX_QUEUES; i++) {
        q = &tcp_pq[i];
        if(q->state != QUEUE_STATE_FREE) {
            clear_tcp_queue(i);
            q->state = QUEUE_STATE_FREE;
        }
        xfree(q->tab);
    }

    close(sock);
    sem_destroy(&send_sem);
}

/*----------------------------------------------------------------------
 * send_packet -  sends a packet via the raw network socket.
 *----------------------------------------------------------------------
 */
int send_packet (Tpacket_data *pd)
{
    struct sockaddr_in dst;
    int len = pd->ipq_pm.data_len;
    int sent;

    dst.sin_family = AF_INET;

    /* Copy destination address from ip header. */
    memcpy(&dst.sin_addr, &pd->ip->daddr, sizeof(pd->ip->daddr));
    
    DPRINT("send_packet: sending packet, len=%u, ip_id=%u.\n", 
           len, ((struct iphdr*)(pd->payload))->id);
    sent = sendto(sock, pd->payload, len, 0, (struct sockaddr *)&dst, sizeof dst);
    if(sent == -1) {
        EPRINT("send_packet: error sending packet.\n");
        perror("send_packet");
    } else if(sent != len) {
        EPRINT("send_packet: sent bytes not equal to packet size: sent=%d\n",
               sent, len);
    }
    dealloc_packet_data(pd);

    return sent;
}

/*----------------------------------------------------------------------
 * queue_packet -  push packet onto selected queue;
 *                 will block until there is space for the packet.
 *----------------------------------------------------------------------
 */
int queue_packet(Tpacket_data *pd, Tpacket_queue* q)
{
    mutex_lock("queue_packet", &sched_mut);

    if(queue_size(q) >= q->max) {  /* Check to see if queue is full. */
        DPRINT("queue_packet: queue is full, dropping packet.\n");
        mutex_unlock("queue_packet", &sched_mut);
        dealloc_packet_data(pd);
        return 1;
    }
        
    q->tab[q->tail] = pd;
    q->tail = (q->tail+1) % q->max;
    queued_pkts++;

    if(DEBUG) {
        int qnum = q-tcp_pq;

        if((qnum >= 0) && (qnum < TCP_MAX_QUEUES)) {
            DPRINT("queue_packet: tcp queue %d, ", qnum); 
        } else if (q == &hipri_pq) {
            DPRINT("queue_packet: queue hipri, ");
        } else if (q == &lowpri_pq) {
            DPRINT("queue_packet: queue lowpri, ");
        } else {
            DPRINT("queue_packet: queue unknown, ");
        }
        DPRINT("size=%d, queued_pkts=%ld\n", queue_size(q), queued_pkts);
    }

    mutex_unlock("queue_packet", &sched_mut);
    sem_post(&send_sem);  /* Indicate there are packets to send.   */

    return 0;
}

/*----------------------------------------------------------------------
 * dequeue_packet -  inline dequeue function, returns NULL on empty;
 *                   assumes sched_mut HELD.
 *----------------------------------------------------------------------
 */
inline Tpacket_data* dequeue_packet(Tpacket_queue* q)
{
    if(queue_empty(q)) {
        return NULL;
    } else {
        Tpacket_data *pd;
        pd = q->tab[q->head];            /* Get packet.                   */
        q->head = (q->head+1) % q->max;  /* Update queue head pointer.    */
        queued_pkts--;                   /* Decrement total packet count. */
        assert(queued_pkts >= 0);
        return pd;    
    }
}

/*----------------------------------------------------------------------
 * mh_ack_rcvd -  reduces in-flight counters based on sent records;
 *                called when an ack is received from the mobile host.
 *----------------------------------------------------------------------
 */
void mh_ack_rcvd(unsigned ack_seq, int queue_num)
{
    Tpacket_queue *q = &tcp_pq[queue_num];
    Tpacket_sent *sent;
    unsigned credit;

    mutex_lock("mh_ack_rcvd", &sched_mut);
       
    credit = 0;
    while(q->sent_hd != NULL && seq_cmp(ack_seq, q->sent_hd->seq) >= 1) {
        sent = q->sent_hd;
        credit += sent->size;
        q->sent_hd = sent->next;
        sent->size = 0;
        /* If the timer is successfully cleared (i.e. it is not currently being
         * activated), then free sent. Otherwise, sent will be freed by
         * packet_timeout. */
        if(clear_timer(packet_timeout, sent) != -1) {
            xfree(sent);
        }
    }
    if(q->sent_hd != NULL)
        q->sent_hd->prev = NULL;
    
    assert(credit <= q->in_flight);
    q->in_flight -= credit;
    tcp_in_flight -= credit;
    DPRINT("mh_ack_rcvd: ack_seq=%u, queue=%d, credit=%u, new tcp_in_flight=%u(%u)\n",
           ack_seq, queue_num, credit, tcp_in_flight, q->in_flight);
    
    check_dealloc_tcp_queue(queue_num);  /* Dealloc closed queue if possible. */
    
    mutex_unlock("mh_ack_rcvd", &sched_mut);    
    sem_post(&send_sem);
}

/*----------------------------------------------------------------------
 * queue_tcp -  queue a tcp packet to the specified tcp queue number.
 *----------------------------------------------------------------------
 */
int queue_tcp(Tpacket_data *pd, int queue_num)
{
    Tpacket_queue *q = &tcp_pq[queue_num];
    struct tcphdr *tcp = pd->trans.tcp;
    
    assert(queue_num >= 0 && queue_num < TCP_MAX_QUEUES);
    
    if(pd->dir == PACKET_DIR_IN) {
        if(tcp->syn) { /* Put packets with SYN set into high priority queue. */
            /* SYN count as 1 byte in flight, but we can ignore this. */
            return queue_packet(pd, &hipri_pq);
        } else {
            return queue_packet(pd, q);
        }
    } else if(pd->dir == PACKET_DIR_OUT) {
        DPRINT("queue_tcp: packet dir=out, sending immediately\n");
        send_packet(pd);
        return 0;
    } else {
        assert(UNREACHABLE);
    }
}

/*----------------------------------------------------------------------
 * queue_hipri -  queue packet as hipri; mainly for ICMP.
 *----------------------------------------------------------------------
 */
int queue_hipri(Tpacket_data *pd)
{
    if(pd->dir == PACKET_DIR_IN) {
        return queue_packet(pd, &hipri_pq);
    } else if(pd->dir == PACKET_DIR_OUT) {
        DPRINT("queue_hipri: packet dir=out, sending immediately\n");
        send_packet(pd);
        return 0;
    } else {
        assert(UNREACHABLE);
    }
}

/*----------------------------------------------------------------------
 * queue_lowpri -  queue packet as lowpri; mainly for UDP.
 *----------------------------------------------------------------------
 */
int queue_lowpri(Tpacket_data *pd)
{
    if(pd->dir == PACKET_DIR_IN) {
        return queue_packet(pd, &lowpri_pq);
    } else if(pd->dir == PACKET_DIR_OUT) {
        DPRINT("queue_lowpri: packet dir=out, sending immediately\n");
        send_packet(pd);
        return 0;
    } else {
        assert(UNREACHABLE);
    }
}

/*----------------------------------------------------------------------
 * scheduler -  entry point of scheduler thread;
 *              packets from TCP queues are transmitted whilst the link
 *              isn't full using token bucket and deficit round robin
 *              algorithms; when there are no packets or the link is full
 *              the scheduler waits on sched_mut.
 *----------------------------------------------------------------------
 */
void scheduler (void *sig_mask)
{
    register Tpacket_queue *q;
    Tpacket_data *pd;
    unsigned tcp_size;  /* Size of TCP data in a packet.               */
    int qnum;           /* Current TCP queue being scheduled.          */
    int wait_on_link;   /* Boolean: link is full, wait on link space.  */
    unsigned curr_dc;   /* Deficit counter for the current TCP queue;
                           zero unless we have to wait for link space. */
    
    if(pthread_sigmask(SIG_BLOCK, sig_mask, NULL)) {
        EPRINT("scheduler: error blocking signals.\n");
    }
    
    qnum = 0;
    curr_dc = 0;
    
    while(!request_exit) {
        if(wait_on_link || queued_pkts == 0) {
            sem_wait(&send_sem);
        }
        wait_on_link = 0;
        
        mutex_lock("scheduler", &sched_mut);
        
        /* Send from high priority queue ASAP. */
        while((pd = dequeue_packet(&hipri_pq)) != NULL) {
            send_packet(pd);
            DPRINT("scheduler: queue hipri, size=%d, queued_pkts=%ld\n",
                   queue_size(&hipri_pq), queued_pkts);
        }
        
        if(qnum >= TCP_MAX_QUEUES) {
            qnum = 0;
        }
        for(; qnum<TCP_MAX_QUEUES; qnum++) {
            q = &tcp_pq[qnum];
            
            if(q->state == QUEUE_STATE_FREE) {
                continue;
                
            } else if(!queue_empty(q)) {
                if(curr_dc == 0) {                  
                    curr_dc = q->dc;
                    curr_dc += DRR_QUANTUM * q->weight;
                }
                while(!queue_empty(q)) {
                    pd = q->tab[q->head];  /* Get packet. */
                    tcp_size = tcp_data_size(pd);
                    if(curr_dc < tcp_size) {
                        break;
                    }
                    if(tcp_in_flight + tcp_size > max_tcp_in_flight) {
                        wait_on_link = 1;
                        break;  /* Go and wait on send_sem if link is full. */
                    }
                    if(tcp_size > 0) {
                        q->in_flight += tcp_size;
                        tcp_in_flight += tcp_size;
                        curr_dc -= tcp_size;
                        add_tcp_sent(pd, tcp_size, q);              
                    }
                    q->head = (q->head+1) % q->max;  /* Update queue head pointer. */
                    queued_pkts--;                   /* Decrement total packet count. */
                    assert(queued_pkts >= 0);
                    
                    send_packet(pd);
                    
                    DPRINT("scheduler: queue %d, size=%d, tcp_in_flight=%u(%u), "
                           "dc=%u, queued_pkts=%ld\n",
                           qnum, queue_size(q), tcp_in_flight, q->in_flight, curr_dc, queued_pkts);
                }
                
                if(wait_on_link) {
                    DPRINT("scheduler: queue %d, link full; will wait on send_sem.\n", qnum);
                    break;
                    
                } else {
                    if(queue_empty(q))
                        q->dc = 0;
                    else
                        q->dc = curr_dc;
                    
                    curr_dc = 0;
                }
            }
            check_dealloc_tcp_queue(qnum);
        }
        
        if(!queue_empty(&lowpri_pq)) {
            /* Send low priority packets only if the link isn't full. */
            if(tcp_in_flight < max_tcp_in_flight) {
                while((pd = dequeue_packet(&lowpri_pq)) != NULL) {
                    send_packet(pd);
                    DPRINT("scheduler: queue lowpri, size=%d, queued_pkts=%ld\n",
                           queue_size(&lowpri_pq), queued_pkts);
                }
            } else {
                wait_on_link = 1;
            }
        }
        
        mutex_unlock("scheduler", &sched_mut);
    }
}

---