net/cor/snd.c

   1 /*
   2  *      Connection oriented routing
   3  *      Copyright (C) 2007-2008 Michael Blizek
   4  *
   5  *      This program is free software; you can redistribute it and/or
   6  *      modify it under the terms of the GNU General Public License
   7  *      as published by the Free Software Foundation; either version 2
   8  *      of the License, or (at your option) any later version.
   9  *
  10  *      This program is distributed in the hope that it will be useful,
  11  *      but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  *      GNU General Public License for more details.
  14  *
  15  *      You should have received a copy of the GNU General Public License
  16  *      along with this program; if not, write to the Free Software
  17  *      Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  18  *      02110-1301, USA.
  19  */
  20
  21 #include <linux/gfp.h>
  22 #include <linux/jiffies.h>
  23 #include <linux/slab.h>
  24
  25 #include "cor.h"
  26
  27 static struct htable retransmits;
  28
  29 static void free_skb(struct ref_counter *cnt)
  30 {
  31         struct skb_procstate *ps = container_of(cnt, struct skb_procstate,
  32                         funcstate.retransmit_queue.refs);
  33
  34         struct sk_buff *skb = skb_from_pstate(ps);
  35
  36         ref_counter_decr(&(ps->rconn->refs));
  37         kfree_skb(skb);
  38 }
  39
  40 static struct ref_counter_def skb_refcnt = {
  41         .free = free_skb
  42 };
  43
  44 struct retransmit_matchparam {
  45         __u32 conn_id;
  46         __u32 seqno;
  47         struct neighbor *nb;
  48 };
  49
  50 static __u32 rm_to_key(struct retransmit_matchparam *rm)
  51 {
  52         return rm->conn_id ^ rm->seqno;
  53 }
  54
  55 /* static struct sk_buff * cor_dequeue(struct Qdisc *sch)
  56 {
  57         struct sk_buff *ret;
  58
  59         struct cor_sched_data *q = qdisc_priv(sch);
  60
  61         struct list_head *ln = q->conn_list.next;
  62         struct conn *best = 0;
  63
  64         __u64 currcost_limit = 0;
  65         __u64 currcost = 0;
  66
  67         spin_lock(&(q->lock));
  68
  69         if (!(skb_queue_empty(&(q->requeue_queue)))) {
  70                 ret = __skb_dequeue(&(q->requeue_queue));
  71                 goto out;
  72         }
  73
  74         while (&(q->conn_list) != ln) {
  75                 __u32 max1, max2, maxcost;
  76                 struct conn *curr = (struct conn *)
  77                                 (((char *) ln) - offsetof(struct conn,
  78                                                 target.out.queue_list));
  79
  80                 BUG_ON(TARGET_OUT != curr->targettype);
  81                 max1 = (256 * ((__u64)curr->credits)) /
  82                                 ((__u64)curr->bytes_queued + curr->avg_rate);
  83
  84                 max2 = (256 * ((__u64)curr->credits +
  85                                 curr->credit_sender - curr->credit_recp)) /
  86                                 ((__u64)curr->bytes_queued + 2*curr->avg_rate);
  87
  88                 maxcost = max((__u32) 0, min((max1), (max2)));
  89
  90                 if (maxcost > currcost_limit) {
  91                         currcost = currcost_limit;
  92                         currcost_limit = maxcost;
  93                         best = curr;
  94                 }
  95
  96                 ln = ln->next;
  97         }
  98
  99         best->credits -= currcost;
 100
 101         ret = __skb_dequeue(&(best->target.out.queue));
 102
 103         if (skb_queue_empty(&(best->target.out.queue))) {
 104                 list_del(&(best->target.out.queue_list));
 105                 best->target.out.qdisc_active = 0;
 106         }
 107
 108 out:
 109         spin_unlock(&(q->lock));
 110
 111         if (likely(0 != ret)) {
 112                 sch->qstats.backlog -= ret->len;
 113                 sch->q.qlen--;
 114         }
 115
 116         return ret;
 117 }
 118
 119 static int cor_enqueue(struct sk_buff *skb, struct Qdisc *sch)
 120 {
 121         struct cor_sched_data *q = qdisc_priv(sch);
 122         struct conn *rconn;
 123
 124         rconn = skb_pstate(skb)->rconn;
 125
 126         BUG_ON(TARGET_OUT != rconn->targettype);
 127
 128         spin_lock(&(rconn->target.out.qdisc_lock));
 129
 130         __skb_queue_tail(&(rconn->target.out.queue), skb);
 131
 132         if (unlikely(0 == rconn->target.out.qdisc_active)) {
 133                 spin_lock(&(q->lock));
 134                 list_add(&(rconn->target.out.queue_list), &(q->conn_list));
 135                 rconn->target.out.qdisc_active = 1;
 136                 spin_unlock(&(q->lock));
 137         }
 138
 139         spin_unlock(&(rconn->target.out.qdisc_lock));
 140
 141         sch->bstats.bytes += skb->len;
 142         sch->bstats.packets++;
 143         sch->q.qlen++;
 144
 145         return NET_XMIT_SUCCESS;
 146 } */
 147
 148 static void cor_xmit(struct sk_buff *skb)
 149 {
 150         struct sk_buff *skb2;
 151
 152         BUG_ON(skb == 0);
 153
 154         skb2 = skb_clone(skb, __GFP_DMA | GFP_KERNEL);
 155
 156         if (skb2 == 0) {
 157                 printk(KERN_WARNING "cor_xmit: cannot clone skb, "
 158                                 "allocation failure?");
 159         }
 160
 161         dev_queue_xmit(skb2);
 162 }
 163
 164 void retransmit_timerfunc(unsigned long arg)
 165 {
 166         unsigned long iflags;
 167
 168         struct neighbor *nb = (struct neighbor *) arg;
 169         struct sk_buff *skb = 0;
 170         struct skb_procstate *ps = 0;
 171         unsigned long timeout;
 172
 173
 174         while (1) {
 175                 spin_lock_irqsave( &(nb->retrans_lock), iflags );
 176                 skb = __skb_dequeue(&(nb->retrans_list));
 177
 178                 if (0 == skb)
 179                         goto out;
 180
 181                 ps = skb_pstate(skb);
 182                 timeout = ps->funcstate.retransmit_queue.timeout;
 183
 184                 if (time_before(timeout, jiffies)) {
 185                         __skb_queue_head(&(nb->retrans_list), skb);
 186                         goto modtimer;
 187                 }
 188
 189                 ps->funcstate.retransmit_queue.timeout = jiffies + nb->latency;
 190                 __skb_queue_tail(&(nb->retrans_list), skb);
 191
 192                 spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 193                 cor_xmit(skb);
 194         }
 195
 196 modtimer:
 197         mod_timer(&(nb->retrans_timer), timeout);
 198
 199 out:
 200         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 201 }
 202
 203 static struct sk_buff *create_packet(struct neighbor *nb, int size,
 204                 gfp_t alloc_flags, __u32 keyid, __u32 keyseq, __u32 conn_id,
 205                 __u32 seqno)
 206 {
 207         struct net_device *dev = nb->dev;
 208         struct sk_buff *ret;
 209         struct skb_procstate *ps;
 210         char *dest;
 211
 212         ret = alloc_skb(size + 17 + LL_ALLOCATED_SPACE(dev), alloc_flags);
 213         if (unlikely(0 == ret))
 214                 return 0;
 215
 216         ret->protocol = htons(ETH_P_COR);
 217         ps = skb_pstate(ret);
 218         ps->funcstate.retransmit_queue.conn_id = conn_id;
 219         ps->funcstate.retransmit_queue.seqno = seqno;
 220         skb_reserve(ret, LL_RESERVED_SPACE(dev));
 221         if(unlikely(dev_hard_header(ret, dev, ETH_P_COR, nb->mac,
 222                         dev->dev_addr, ret->len) < 0))
 223                 return 0;
 224
 225         skb_reserve(ret, LL_RESERVED_SPACE(dev));
 226         skb_reset_network_header(ret);
 227
 228         dest = skb_put(ret, 17);
 229         BUG_ON(0 == dest);
 230
 231         dest[0] = PACKET_TYPE_DATA;
 232         dest += 1;
 233
 234         put_u32(dest, keyid, 1);
 235         dest += 4;
 236         put_u32(dest, keyseq, 1);
 237         dest += 4;
 238         put_u32(dest, conn_id, 1);
 239         dest += 4;
 240         put_u32(dest, seqno, 1);
 241         dest += 4;
 242
 243         return ret;
 244 }
 245
 246 struct sk_buff *create_packet_conn(struct conn *target, int size,
 247                 gfp_t alloc_flags)
 248 {
 249         __u32 connid = target->target.out.conn_id;
 250         __u32 seqno;
 251
 252         seqno = target->target.out.seqno;
 253         target->target.out.seqno += size;
 254
 255         return create_packet(target->target.out.nb, size, alloc_flags, 0, 0,
 256                         connid, seqno);
 257 }
 258
 259 struct sk_buff *create_packet_kernel(struct neighbor *nb, int size,
 260                 gfp_t alloc_flags)
 261 {
 262         __u32 seqno = atomic_add_return(1, &(nb->kpacket_seqno));
 263         return create_packet(nb, size, alloc_flags, 0, 0, 0, seqno);
 264 }
 265
 266 void send_conn_flushdata(struct conn *rconn, char *data, __u32 datalen)
 267 {
 268         __u32 seqno;
 269         struct control_msg_out *cm = alloc_control_msg();
 270
 271         seqno = rconn->target.out.seqno;
 272         rconn->target.out.seqno += datalen;
 273
 274         #warning todo retransmit/controlmsg == 0
 275
 276         send_conndata(cm, rconn->target.out.nb, rconn->target.out.conn_id,
 277                         seqno, data, data, datalen);
 278 }
 279
 280 void send_packet(struct sk_buff *skb, struct neighbor *nb)
 281 {
 282         unsigned long iflags;
 283
 284         struct skb_procstate *ps = skb_pstate(skb);
 285         struct retransmit_matchparam rm;
 286         int first;
 287
 288         printk(KERN_ERR "send_packet");
 289
 290         BUG_ON(ps->rconn->targettype != TARGET_OUT);
 291
 292         rm.conn_id = ps->funcstate.retransmit_queue.conn_id;
 293         rm.seqno = ps->funcstate.retransmit_queue.seqno;
 294         rm.nb = nb;
 295
 296         ps->funcstate.retransmit_queue.timeout = jiffies + nb->latency;
 297         ref_counter_init(&(ps->funcstate.retransmit_queue.refs), &skb_refcnt);
 298
 299         htable_insert(&retransmits, (char *) skb, rm_to_key(&rm));
 300
 301         spin_lock_irqsave( &(nb->retrans_lock), iflags );
 302
 303         first = unlikely(skb_queue_empty(&(nb->retrans_list)));
 304         __skb_queue_tail(&(nb->retrans_list), skb);
 305
 306         if (first) {
 307                 mod_timer(&(nb->retrans_timer),
 308                                 ps->funcstate.retransmit_queue.timeout);
 309         }
 310
 311         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 312
 313         ref_counter_incr(&(ps->rconn->refs));
 314
 315         cor_xmit(skb);
 316
 317         ref_counter_incr(&(ps->funcstate.retransmit_queue.refs));
 318 }
 319
 320 void ack_received(struct neighbor *nb, __u32 conn_id, __u32 seqno,
 321                 int nack)
 322 {
 323         unsigned long iflags;
 324
 325         struct sk_buff *skb = 0;
 326         struct skb_procstate *ps;
 327         struct retransmit_matchparam rm;
 328
 329         int first = 0;
 330
 331         int ret;
 332
 333         rm.conn_id = conn_id;
 334         rm.seqno = seqno;
 335         rm.nb = nb;
 336
 337         skb = (struct sk_buff *) htable_get(&retransmits, rm_to_key(&rm), &rm);
 338
 339         if (0 == skb)
 340                 return;
 341
 342         ps = skb_pstate(skb);
 343
 344         ret = htable_delete(&retransmits, rm_to_key(&rm), &rm);
 345         if (ret) {
 346                  /* somebody else has already deleted it in the meantime */
 347                 return;
 348         }
 349
 350         spin_lock_irqsave( &(nb->retrans_lock), iflags );
 351
 352         if (unlikely(nb->retrans_list.next == skb))
 353                 first = 1;
 354         skb->next->prev = skb->prev;
 355         skb->prev->next = skb->next;
 356
 357         if (first) {
 358                 if (unlikely(skb_queue_empty(&(nb->retrans_list)))) {
 359                         mod_timer(&(nb->retrans_timer), jiffies + nb->latency);
 360                 }
 361         }
 362
 363         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 364 }
 365
 366 void flush_out(struct conn *rconn)
 367 {
 368         int targetmss = mss(rconn->target.out.nb);
 369
 370         char *buf;
 371         __u32 len;
 372
 373         printk(KERN_ERR "flush_out");
 374
 375         if (unlikely(rconn->target.out.conn_id == 0))
 376                 return;
 377
 378         while (rconn->buf.read_remaining >= targetmss) {
 379                 struct sk_buff *newskb = create_packet_conn(rconn, targetmss,
 380                                 GFP_ATOMIC);
 381
 382                 char *dst = skb_put(newskb, targetmss);
 383                 databuf_pull(&(rconn->buf), dst, targetmss);
 384                 databuf_ackread(&(rconn->buf));
 385                 send_packet(newskb, rconn->target.out.nb);
 386         }
 387
 388         if (rconn->buf.read_remaining == 0)
 389                 return;
 390
 391         len = rconn->buf.read_remaining;
 392         buf = kmalloc(len, GFP_KERNEL);
 393
 394         databuf_pull(&(rconn->buf), buf, len);
 395         databuf_ackread(&(rconn->buf));
 396
 397         send_conn_flushdata(rconn, buf, len);
 398
 399         wake_sender(rconn);
 400         printk(KERN_ERR "flush_out1");
 401 }
 402
 403 static int matches_skb_connid_seqno(void *htentry, void *searcheditem)
 404 {
 405         struct sk_buff *skb = (struct sk_buff *) htentry;
 406         struct skb_procstate *ps = skb_pstate(skb);
 407         struct retransmit_matchparam *rm = (struct retransmit_matchparam *)
 408                         searcheditem;
 409
 410         return rm->conn_id == ps->funcstate.retransmit_queue.conn_id &&
 411                         rm->seqno == ps->funcstate.retransmit_queue.seqno &&
 412                         rm->nb == ps->rconn->target.out.nb;
 413 }
 414
 415 static inline __u32 retransmit_entryoffset(void)
 416 {
 417         return offsetof(struct sk_buff, cb) + offsetof(struct skb_procstate,
 418                         funcstate.retransmit_queue.htab_entry);
 419 }
 420
 421 static inline __u32 retransmit_refsoffset(void)
 422 {
 423         return offsetof(struct sk_buff, cb) + offsetof(struct skb_procstate,
 424                         funcstate.retransmit_queue.refs);
 425 }
 426
 427 int __init cor_snd_init(void)
 428 {
 429         htable_init(&retransmits, matches_skb_connid_seqno,
 430                 retransmit_entryoffset(), retransmit_refsoffset());
 431
 432         return 0;
 433 }
 434
 435 MODULE_LICENSE("GPL");