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, int atomic)
 149 {
 150         struct sk_buff *skb2;
 151
 152         BUG_ON(skb == 0);
 153
 154         skb2 = skb_clone(skb, __GFP_DMA | (atomic ? GFP_ATOMIC : GFP_KERNEL));
 155
 156         if (skb2 == 0) {
 157                 printk(KERN_WARNING "cor_xmit: cannot clone skb, "
 158                                 "allocation failure?");
 159                 return;
 160         }
 161
 162         printk(KERN_ERR "xmit");
 163         dev_queue_xmit(skb2);
 164 }
 165
 166 static void set_retrans_timeout(struct sk_buff *skb, struct neighbor *nb)
 167 {
 168         struct skb_procstate *ps = skb_pstate(skb);
 169         ps->funcstate.retransmit_queue.timeout = jiffies +
 170                         msecs_to_jiffies(300 + nb->latency);
 171 }
 172
 173 void retransmit_timerfunc(unsigned long arg)
 174 {
 175         unsigned long iflags;
 176
 177         struct neighbor *nb = (struct neighbor *) arg;
 178         struct sk_buff *skb = 0;
 179         unsigned long timeout;
 180
 181         while (1) {
 182                 spin_lock_irqsave( &(nb->retrans_lock), iflags );
 183                 skb = __skb_dequeue(&(nb->retrans_list));
 184
 185                 if (0 == skb)
 186                         goto out;
 187
 188                 timeout = skb_pstate(skb)->funcstate.retransmit_queue.timeout;
 189
 190                 if (time_after(timeout, jiffies)) {
 191                         __skb_queue_head(&(nb->retrans_list), skb);
 192                         goto modtimer;
 193                 }
 194
 195                 set_retrans_timeout(skb, nb);
 196
 197                 __skb_queue_tail(&(nb->retrans_list), skb);
 198
 199                 spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 200                 cor_xmit(skb, 1);
 201         }
 202
 203 modtimer:
 204         mod_timer(&(nb->retrans_timer), timeout);
 205
 206 out:
 207         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 208 }
 209
 210 static struct sk_buff *create_packet(struct neighbor *nb, int size,
 211                 gfp_t alloc_flags, __u32 conn_id, __u32 seqno,
 212                 struct conn *target)
 213 {
 214         struct net_device *dev = nb->dev;
 215         struct sk_buff *ret;
 216         struct skb_procstate *ps;
 217         char *dest;
 218
 219         ret = alloc_skb(size + 9 + LL_ALLOCATED_SPACE(dev), alloc_flags);
 220         if (unlikely(0 == ret))
 221                 return 0;
 222
 223         ret->protocol = htons(ETH_P_COR);
 224         ret->dev = dev;
 225
 226         skb_reserve(ret, LL_RESERVED_SPACE(dev));
 227         if(unlikely(dev_hard_header(ret, dev, ETH_P_COR, nb->mac,
 228                         dev->dev_addr, ret->len) < 0))
 229                 return 0;
 230         skb_reset_network_header(ret);
 231
 232         ps = skb_pstate(ret);
 233         ps->rconn = target;
 234         memset(&(ps->funcstate.retransmit_queue), 0,
 235                         sizeof(ps->funcstate.retransmit_queue));
 236         #warning todo funcstate.retransmit_queue.htab_entry
 237         ps->funcstate.retransmit_queue.conn_id = conn_id;
 238         ps->funcstate.retransmit_queue.seqno = seqno;
 239         ps->funcstate.retransmit_queue.nb = nb;
 240         #warning todo decr refcnt
 241         ref_counter_incr(&(nb->refs));
 242
 243         dest = skb_put(ret, 9);
 244         BUG_ON(0 == dest);
 245
 246         dest[0] = PACKET_TYPE_DATA;
 247         dest += 1;
 248
 249         put_u32(dest, conn_id, 1);
 250         dest += 4;
 251         put_u32(dest, seqno, 1);
 252         dest += 4;
 253
 254         return ret;
 255 }
 256
 257 struct sk_buff *create_packet_conn(struct conn *target, int size,
 258                 gfp_t alloc_flags)
 259 {
 260         __u32 connid = target->target.out.conn_id;
 261         __u32 seqno;
 262
 263         BUG_ON(target->targettype != TARGET_OUT);
 264
 265         seqno = target->target.out.seqno;
 266         target->target.out.seqno += size;
 267
 268         return create_packet(target->target.out.nb, size, alloc_flags,
 269                         connid, seqno, target);
 270 }
 271
 272 struct sk_buff *create_packet_kernel(struct neighbor *nb, int size,
 273                 gfp_t alloc_flags)
 274 {
 275         __u32 seqno = atomic_add_return(1, &(nb->kpacket_seqno));
 276         return create_packet(nb, size, alloc_flags, 0, seqno, 0);
 277 }
 278
 279 void send_conn_flushdata(struct conn *rconn, char *data, __u32 datalen)
 280 {
 281         __u32 seqno;
 282         struct control_msg_out *cm = alloc_control_msg();
 283
 284         seqno = rconn->target.out.seqno;
 285         rconn->target.out.seqno += datalen;
 286
 287         #warning todo retransmit/controlmsg == 0
 288
 289         send_conndata(cm, rconn->target.out.nb, rconn->target.out.conn_id,
 290                         seqno, data, data, datalen);
 291 }
 292
 293 static void schedule_retransmit(struct sk_buff *skb, struct neighbor *nb)
 294 {
 295         unsigned long iflags;
 296
 297         struct skb_procstate *ps = skb_pstate(skb);
 298         struct retransmit_matchparam rm;
 299         int first;
 300
 301         rm.conn_id = ps->funcstate.retransmit_queue.conn_id;
 302         rm.seqno = ps->funcstate.retransmit_queue.seqno;
 303         rm.nb = nb;
 304
 305         set_retrans_timeout(skb, nb);
 306         ref_counter_init(&(ps->funcstate.retransmit_queue.refs), &skb_refcnt);
 307         htable_insert(&retransmits, (char *) skb, rm_to_key(&rm));
 308         spin_lock_irqsave( &(nb->retrans_lock), iflags );
 309         first = unlikely(skb_queue_empty(&(nb->retrans_list)));
 310         __skb_queue_tail(&(nb->retrans_list), skb);
 311
 312         if (first) {
 313                 mod_timer(&(nb->retrans_timer),
 314                                 ps->funcstate.retransmit_queue.timeout);
 315         }
 316
 317         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 318
 319         ref_counter_incr(&(ps->funcstate.retransmit_queue.refs));
 320 }
 321
 322 void send_packet(struct sk_buff *skb, struct neighbor *nb, int retransmit)
 323 {
 324         struct skb_procstate *ps = skb_pstate(skb);
 325
 326         if (retransmit)
 327                 schedule_retransmit(skb, nb);
 328
 329         if (ps->rconn != 0)
 330                 ref_counter_incr(&(ps->rconn->refs));
 331
 332         cor_xmit(skb, 0);
 333 }
 334
 335 void ack_received(struct neighbor *nb, __u32 conn_id, __u32 seqno)
 336 {
 337         unsigned long iflags;
 338
 339         struct sk_buff *skb = 0;
 340         struct skb_procstate *ps;
 341         struct retransmit_matchparam rm;
 342
 343         int first = 0;
 344
 345         int ret;
 346
 347         rm.conn_id = conn_id;
 348         rm.seqno = seqno;
 349         rm.nb = nb;
 350
 351         skb = (struct sk_buff *) htable_get(&retransmits, rm_to_key(&rm), &rm);
 352
 353         if (0 == skb) {
 354                 printk(KERN_ERR "bogus/duplicate ack received");
 355                 return;
 356         }
 357
 358         ps = skb_pstate(skb);
 359
 360         ret = htable_delete(&retransmits, rm_to_key(&rm), &rm);
 361         if (ret) {
 362                  /* somebody else has already deleted it in the meantime */
 363                 return;
 364         }
 365
 366         spin_lock_irqsave( &(nb->retrans_lock), iflags );
 367
 368         if (unlikely(nb->retrans_list.next == skb))
 369                 first = 1;
 370         skb->next->prev = skb->prev;
 371         skb->prev->next = skb->next;
 372
 373         if (first) {
 374                 if (unlikely(skb_queue_empty(&(nb->retrans_list)))) {
 375                         mod_timer(&(nb->retrans_timer), jiffies + nb->latency);
 376                 }
 377         }
 378
 379         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 380 }
 381
 382 void flush_out(struct conn *rconn)
 383 {
 384         int targetmss = mss(rconn->target.out.nb);
 385
 386         char *buf;
 387         __u32 len;
 388
 389         if (unlikely(rconn->target.out.conn_id == 0))
 390                 return;
 391
 392         while (rconn->buf.read_remaining >= targetmss) {
 393                 struct sk_buff *newskb = create_packet_conn(rconn, targetmss,
 394                                 GFP_ATOMIC);
 395
 396                 char *dst = skb_put(newskb, targetmss);
 397                 databuf_pull(&(rconn->buf), dst, targetmss);
 398                 databuf_ackread(&(rconn->buf));
 399                 send_packet(newskb, rconn->target.out.nb, 1);
 400         }
 401
 402         if (rconn->buf.read_remaining == 0)
 403                 return;
 404
 405         len = rconn->buf.read_remaining;
 406         buf = kmalloc(len, GFP_KERNEL);
 407
 408         databuf_pull(&(rconn->buf), buf, len);
 409         databuf_ackread(&(rconn->buf));
 410
 411         send_conn_flushdata(rconn, buf, len);
 412
 413         wake_sender(rconn);
 414 }
 415
 416 static int matches_skb_connid_seqno(void *htentry, void *searcheditem)
 417 {
 418         struct sk_buff *skb = (struct sk_buff *) htentry;
 419         struct skb_procstate *ps = skb_pstate(skb);
 420         struct retransmit_matchparam *rm = (struct retransmit_matchparam *)
 421                         searcheditem;
 422
 423         return rm->conn_id == ps->funcstate.retransmit_queue.conn_id &&
 424                         rm->seqno == ps->funcstate.retransmit_queue.seqno &&
 425                         rm->nb == ps->funcstate.retransmit_queue.nb;
 426 }
 427
 428 static inline __u32 retransmit_entryoffset(void)
 429 {
 430         return offsetof(struct sk_buff, cb) + offsetof(struct skb_procstate,
 431                         funcstate.retransmit_queue.htab_entry);
 432 }
 433
 434 static inline __u32 retransmit_refsoffset(void)
 435 {
 436         return offsetof(struct sk_buff, cb) + offsetof(struct skb_procstate,
 437                         funcstate.retransmit_queue.refs);
 438 }
 439
 440 int __init cor_snd_init(void)
 441 {
 442         htable_init(&retransmits, matches_skb_connid_seqno,
 443                 retransmit_entryoffset(), retransmit_refsoffset());
 444
 445         return 0;
 446 }
 447
 448 MODULE_LICENSE("GPL");