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 atomic_t kpacket_seqno = ATOMIC_INIT(1);
  30
  31 static void free_skb(struct ref_counter *cnt)
  32 {
  33         struct skb_procstate *ps = container_of(cnt, struct skb_procstate,
  34                         funcstate.retransmit_queue.refs);
  35
  36         struct sk_buff *skb = skb_from_pstate(ps);
  37
  38         ref_counter_decr(&(ps->rconn->refs));
  39         kfree_skb(skb);
  40 }
  41
  42 static struct ref_counter_def skb_refcnt = {
  43         .free = free_skb
  44 };
  45
  46 struct retransmit_matchparam {
  47         __u32 conn_id;
  48         __u32 seqno;
  49         struct neighbor *nb;
  50 };
  51
  52 static __u32 rm_to_key(struct retransmit_matchparam *rm)
  53 {
  54         return rm->conn_id ^ rm->seqno;
  55 }
  56
  57 /* static struct sk_buff * cor_dequeue(struct Qdisc *sch)
  58 {
  59         struct sk_buff *ret;
  60
  61         struct cor_sched_data *q = qdisc_priv(sch);
  62
  63         struct list_head *ln = q->conn_list.next;
  64         struct conn *best = 0;
  65
  66         __u64 currcost_limit = 0;
  67         __u64 currcost = 0;
  68
  69         spin_lock(&(q->lock));
  70
  71         if (!(skb_queue_empty(&(q->requeue_queue)))) {
  72                 ret = __skb_dequeue(&(q->requeue_queue));
  73                 goto out;
  74         }
  75
  76         while (&(q->conn_list) != ln) {
  77                 __u32 max1, max2, maxcost;
  78                 struct conn *curr = (struct conn *)
  79                                 (((char *) ln) - offsetof(struct conn,
  80                                                 target.out.queue_list));
  81
  82                 BUG_ON(TARGET_OUT != curr->targettype);
  83                 max1 = (256 * ((__u64)curr->credits)) /
  84                                 ((__u64)curr->bytes_queued + curr->avg_rate);
  85
  86                 max2 = (256 * ((__u64)curr->credits +
  87                                 curr->credit_sender - curr->credit_recp)) /
  88                                 ((__u64)curr->bytes_queued + 2*curr->avg_rate);
  89
  90                 maxcost = max((__u32) 0, min((max1), (max2)));
  91
  92                 if (maxcost > currcost_limit) {
  93                         currcost = currcost_limit;
  94                         currcost_limit = maxcost;
  95                         best = curr;
  96                 }
  97
  98                 ln = ln->next;
  99         }
 100
 101         best->credits -= currcost;
 102
 103         ret = __skb_dequeue(&(best->target.out.queue));
 104
 105         if (skb_queue_empty(&(best->target.out.queue))) {
 106                 list_del(&(best->target.out.queue_list));
 107                 best->target.out.qdisc_active = 0;
 108         }
 109
 110 out:
 111         spin_unlock(&(q->lock));
 112
 113         if (likely(0 != ret)) {
 114                 sch->qstats.backlog -= ret->len;
 115                 sch->q.qlen--;
 116         }
 117
 118         return ret;
 119 }
 120
 121 static int cor_enqueue(struct sk_buff *skb, struct Qdisc *sch)
 122 {
 123         struct cor_sched_data *q = qdisc_priv(sch);
 124         struct conn *rconn;
 125
 126         rconn = skb_pstate(skb)->rconn;
 127
 128         BUG_ON(TARGET_OUT != rconn->targettype);
 129
 130         spin_lock(&(rconn->target.out.qdisc_lock));
 131
 132         __skb_queue_tail(&(rconn->target.out.queue), skb);
 133
 134         if (unlikely(0 == rconn->target.out.qdisc_active)) {
 135                 spin_lock(&(q->lock));
 136                 list_add(&(rconn->target.out.queue_list), &(q->conn_list));
 137                 rconn->target.out.qdisc_active = 1;
 138                 spin_unlock(&(q->lock));
 139         }
 140
 141         spin_unlock(&(rconn->target.out.qdisc_lock));
 142
 143         sch->bstats.bytes += skb->len;
 144         sch->bstats.packets++;
 145         sch->q.qlen++;
 146
 147         return NET_XMIT_SUCCESS;
 148 } */
 149
 150 static void cor_xmit(struct sk_buff *skb)
 151 {
 152         struct sk_buff *skb2;
 153
 154         BUG_ON(skb == 0);
 155
 156         skb2 = skb_clone(skb, __GFP_DMA | GFP_KERNEL);
 157
 158         if (skb2 == 0) {
 159                 printk(KERN_WARNING "cor_xmit: cannot clone skb, "
 160                                 "allocation failure?");
 161         }
 162
 163         dev_queue_xmit(skb2);
 164 }
 165
 166 void retransmit_timerfunc(unsigned long arg)
 167 {
 168         unsigned long iflags;
 169
 170         struct neighbor *nb = (struct neighbor *) arg;
 171         struct sk_buff *skb = 0;
 172         struct skb_procstate *ps = 0;
 173         unsigned long timeout;
 174
 175
 176         while (1) {
 177                 spin_lock_irqsave( &(nb->retrans_lock), iflags );
 178                 skb = __skb_dequeue(&(nb->retrans_list));
 179
 180                 if (0 == skb)
 181                         goto out;
 182
 183                 ps = skb_pstate(skb);
 184                 timeout = ps->funcstate.retransmit_queue.timeout;
 185
 186                 if (time_before(timeout, jiffies)) {
 187                         __skb_queue_head(&(nb->retrans_list), skb);
 188                         goto modtimer;
 189                 }
 190
 191                 ps->funcstate.retransmit_queue.timeout = jiffies + nb->latency;
 192                 __skb_queue_tail(&(nb->retrans_list), skb);
 193
 194                 spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 195                 cor_xmit(skb);
 196         }
 197
 198 modtimer:
 199         mod_timer(&(nb->retrans_timer), timeout);
 200
 201 out:
 202         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 203 }
 204
 205 static struct sk_buff *create_packet(struct neighbor *nb, int size,
 206                 gfp_t alloc_flags, __u32 keyid, __u32 keyseq, __u32 conn_id,
 207                 __u32 seqno)
 208 {
 209         struct net_device *dev = nb->dev;
 210         struct sk_buff *ret;
 211         struct skb_procstate *ps;
 212         char *dest;
 213
 214         ret = alloc_skb(size + 17 + LL_ALLOCATED_SPACE(dev), alloc_flags);
 215         if (unlikely(0 == ret))
 216                 return 0;
 217
 218         ret->protocol = htons(ETH_P_COR);
 219         ps = skb_pstate(ret);
 220         ps->funcstate.retransmit_queue.conn_id = conn_id;
 221         ps->funcstate.retransmit_queue.seqno = seqno;
 222         skb_reserve(ret, LL_RESERVED_SPACE(dev));
 223         if(unlikely(dev_hard_header(ret, dev, ETH_P_COR, nb->mac,
 224                         dev->dev_addr, ret->len) < 0))
 225                 return 0;
 226
 227         skb_reserve(ret, LL_RESERVED_SPACE(dev));
 228         skb_reset_network_header(ret);
 229
 230         dest = skb_put(ret, 17);
 231         BUG_ON(0 == dest);
 232
 233         dest[0] = PACKET_TYPE_DATA;
 234         dest += 1;
 235
 236         put_u32(dest, keyid, 1);
 237         dest += 4;
 238         put_u32(dest, keyseq, 1);
 239         dest += 4;
 240         put_u32(dest, conn_id, 1);
 241         dest += 4;
 242         put_u32(dest, seqno, 1);
 243         dest += 4;
 244
 245         return ret;
 246 }
 247
 248 struct sk_buff *create_packet_conn(struct conn *target, int size,
 249                 gfp_t alloc_flags)
 250 {
 251         __u32 connid = target->target.out.conn_id;
 252         __u32 seqno;
 253
 254         seqno = target->target.out.seqno;
 255         target->target.out.seqno += size;
 256
 257         atomic_inc(&(target->target.out.inflight_packets));
 258         return create_packet(target->target.out.nb, size, alloc_flags, 0, 0,
 259                         connid, seqno);
 260 }
 261
 262 struct sk_buff *create_packet_kernel(struct neighbor *nb, int size,
 263                 gfp_t alloc_flags)
 264 {
 265         __u32 seqno = 0;
 266
 267         while (seqno == 0) {
 268                 atomic_add_return(1, &kpacket_seqno);
 269         }
 270         return create_packet(nb, size, alloc_flags, 0, 0, 0, seqno);
 271 }
 272
 273 void send_conn_flushdata(struct conn *rconn, char *data, __u32 datalen)
 274 {
 275         __u32 seqno;
 276
 277         seqno = rconn->target.out.seqno;
 278         rconn->target.out.seqno += datalen;
 279
 280         #warning todo retransmit
 281
 282         send_conndata(rconn->target.out.nb, rconn->target.out.conn_id, seqno,
 283                 data, data, datalen);
 284 }
 285
 286 void send_packet(struct sk_buff *skb, struct neighbor *nb)
 287 {
 288         unsigned long iflags;
 289
 290         struct skb_procstate *ps = skb_pstate(skb);
 291         struct retransmit_matchparam rm;
 292         int first;
 293
 294         BUG_ON(ps->rconn->targettype != TARGET_OUT);
 295
 296         rm.conn_id = ps->funcstate.retransmit_queue.conn_id;
 297         rm.seqno = ps->funcstate.retransmit_queue.seqno;
 298         rm.nb = nb;
 299
 300         ps->funcstate.retransmit_queue.timeout = jiffies + nb->latency;
 301         ref_counter_init(&(ps->funcstate.retransmit_queue.refs), &skb_refcnt);
 302
 303         htable_insert(&retransmits, (char *) skb, rm_to_key(&rm));
 304
 305         spin_lock_irqsave( &(nb->retrans_lock), iflags );
 306
 307         first = unlikely(skb_queue_empty(&(nb->retrans_list)));
 308         __skb_queue_tail(&(nb->retrans_list), skb);
 309
 310         if (first) {
 311                 mod_timer(&(nb->retrans_timer),
 312                                 ps->funcstate.retransmit_queue.timeout);
 313         }
 314
 315         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 316
 317         ref_counter_incr(&(ps->rconn->refs));
 318
 319         cor_xmit(skb);
 320
 321         ref_counter_incr(&(ps->funcstate.retransmit_queue.refs));
 322 }
 323
 324 void ack_received(struct neighbor *nb, __u32 conn_id, __u32 seqno,
 325                 int nack)
 326 {
 327         unsigned long iflags;
 328
 329         struct sk_buff *skb = 0;
 330         struct skb_procstate *ps;
 331         struct retransmit_matchparam rm;
 332
 333         int first = 0;
 334
 335         int ret;
 336
 337         rm.conn_id = conn_id;
 338         rm.seqno = seqno;
 339         rm.nb = nb;
 340
 341         skb = (struct sk_buff *) htable_get(&retransmits, rm_to_key(&rm), &rm);
 342
 343         if (0 == skb)
 344                 return;
 345
 346         ps = skb_pstate(skb);
 347
 348         ret = htable_delete(&retransmits, (char *) skb, rm_to_key(&rm), &rm);
 349         if (ret) {
 350                  /* somebody else has already deleted it in the meantime */
 351                 return;
 352         }
 353
 354         spin_lock_irqsave( &(nb->retrans_lock), iflags );
 355
 356         if (unlikely(nb->retrans_list.next == skb))
 357                 first = 1;
 358         skb->next->prev = skb->prev;
 359         skb->prev->next = skb->next;
 360
 361         if (first) {
 362                 if (unlikely(skb_queue_empty(&(nb->retrans_list)))) {
 363                         mod_timer(&(nb->retrans_timer), jiffies + nb->latency);
 364                 }
 365         }
 366
 367         spin_unlock_irqrestore( &(nb->retrans_lock), iflags );
 368
 369         if (atomic_dec_and_test(&(ps->rconn->target.out.inflight_packets))) {
 370                 if (ps->rconn->sourcetype == SOURCE_IN) {
 371                         mutex_lock(&(ps->rconn->source.in.rcv_lock));
 372                         flush_rbuf(ps->rconn);
 373                         mutex_unlock(&(ps->rconn->source.in.rcv_lock));
 374                 } else if (ps->rconn->sourcetype == SOURCE_SOCK) {
 375                         mutex_lock(&(ps->rconn->source.sock.lock));
 376                         cor_flush_sockbuf(ps->rconn);
 377                         mutex_unlock(&(ps->rconn->source.sock.lock));
 378                 } else {
 379                         BUG();
 380                 }
 381         }
 382 }
 383
 384 static int matches_skb_connid_seqno(void *htentry, void *searcheditem)
 385 {
 386         struct sk_buff *skb = (struct sk_buff *) htentry;
 387         struct skb_procstate *ps = skb_pstate(skb);
 388         struct retransmit_matchparam *rm = (struct retransmit_matchparam *)
 389                         searcheditem;
 390
 391         return rm->conn_id == ps->funcstate.retransmit_queue.conn_id &&
 392                         rm->seqno == ps->funcstate.retransmit_queue.seqno &&
 393                         rm->nb == ps->rconn->target.out.nb;
 394 }
 395
 396 static inline __u32 retransmit_entryoffset(void)
 397 {
 398         return offsetof(struct sk_buff, cb) + offsetof(struct skb_procstate,
 399                         funcstate.retransmit_queue.htab_entry);
 400 }
 401
 402 static inline __u32 retransmit_refsoffset(void)
 403 {
 404         return offsetof(struct sk_buff, cb) + offsetof(struct skb_procstate,
 405                         funcstate.retransmit_queue.refs);
 406 }
 407
 408 int __init cor_snd_init(void)
 409 {
 410         htable_init(&retransmits, matches_skb_connid_seqno,
 411                 retransmit_entryoffset(), retransmit_refsoffset());
 412
 413         return 0;
 414 }
 415
 416 MODULE_LICENSE("GPL");