net/cor/forward.c

   1 /**
   2  *      Connection oriented routing
   3  *      Copyright (C) 2007-2011 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/mutex.h>
  22
  23 #include "cor.h"
  24
  25 struct kmem_cache *data_buf_item_slab;
  26
  27 #define PAGESIZE (1 << PAGE_SHIFT)
  28
  29 void databuf_init(struct conn *cn_init)
  30 {
  31         memset(&(cn_init->data_buf), 0, sizeof(cn_init->data_buf));
  32         INIT_LIST_HEAD(&(cn_init->data_buf.items));
  33 }
  34
  35 static void databuf_item_free(struct conn *cn_l, struct data_buf_item *item)
  36 {
  37         list_del(&(item->buf_list));
  38         if (item->type == TYPE_BUF) {
  39                 cn_l->data_buf.overhead -= sizeof(struct data_buf_item) +
  40                                 item->buflen - item->datalen;
  41                 kfree(item->buf);
  42                 kmem_cache_free(data_buf_item_slab, item);
  43         } else if (item->type == TYPE_SKB) {
  44                 struct sk_buff *skb = skb_from_pstate(container_of(item,
  45                                 struct skb_procstate, funcstate.rcv2.dbi));
  46                 cn_l->data_buf.overhead -= sizeof(struct sk_buff);
  47                 kfree_skb(skb);
  48         } else {
  49                 BUG();
  50         }
  51 }
  52
  53 void databuf_ackdiscard(struct conn *cn_l)
  54 {
  55         __u32 freed = 0;
  56         while (!list_empty(&(cn_l->data_buf.items))) {
  57                 struct data_buf_item *item = container_of(
  58                                 cn_l->data_buf.items.next,
  59                                 struct data_buf_item, buf_list);
  60                 freed += item->datalen;
  61
  62                 databuf_item_free(cn_l, item);
  63         }
  64
  65         cn_l->data_buf.totalsize -= freed;
  66         cn_l->data_buf.first_offset += freed;
  67
  68         BUG_ON(cn_l->data_buf.totalsize != 0);
  69         BUG_ON(cn_l->data_buf.overhead != 0);
  70
  71         if (cn_l->data_buf.cpacket_buffer != 0) {
  72                 free_cpacket_buffer(cn_l->data_buf.cpacket_buffer);
  73                 cn_l->data_buf.cpacket_buffer = 0;
  74         }
  75
  76         cn_l->data_buf.read_remaining = 0;
  77         cn_l->data_buf.last_read_offset = 0;
  78         cn_l->data_buf.lastread = 0;
  79
  80         if (cn_l->isreset == 0 && cn_l->sourcetype == SOURCE_IN)
  81                 refresh_speedstat(cn_l, freed);
  82 }
  83
  84 void reset_seqno(struct conn *cn_l, __u32 initseqno)
  85 {
  86         cn_l->data_buf.first_offset = initseqno -
  87                         cn_l->data_buf.totalsize +
  88                         cn_l->data_buf.read_remaining;
  89 }
  90
  91 static void databuf_nextreadchunk(struct conn *cn_l)
  92 {
  93         if (cn_l->data_buf.lastread == 0) {
  94                 BUG_ON(cn_l->data_buf.last_read_offset != 0);
  95                 BUG_ON(list_empty(&(cn_l->data_buf.items)));
  96                 cn_l->data_buf.lastread = container_of(
  97                                 cn_l->data_buf.items.next,
  98                                 struct data_buf_item, buf_list);
  99         } else if (&(cn_l->data_buf.lastread->buf_list) !=
 100                         cn_l->data_buf.items.prev) {
 101                 cn_l->data_buf.lastread = container_of(
 102                                 cn_l->data_buf.lastread->buf_list.next,
 103                                 struct data_buf_item, buf_list);
 104
 105                 cn_l->data_buf.last_read_offset = 0;
 106         }
 107 }
 108
 109 static int _databuf_pull(struct conn *cn_l, char *dst, int len, int userbuf)
 110 {
 111         int totalcpy = 0;
 112
 113         BUG_ON(cn_l->data_buf.read_remaining < len);
 114
 115         if (cn_l->data_buf.lastread == 0)
 116                 databuf_nextreadchunk(cn_l);
 117
 118         while(len > 0) {
 119                 int rc = 0;
 120                 int cpy = len;
 121
 122                 char *srcbufcpystart = 0;
 123                 int srcbufcpylen = 0;
 124
 125                 BUG_ON(cn_l->data_buf.lastread == 0);
 126
 127                 srcbufcpystart = cn_l->data_buf.lastread->buf +
 128                                 cn_l->data_buf.last_read_offset;
 129                 srcbufcpylen = cn_l->data_buf.lastread->datalen -
 130                                 cn_l->data_buf.last_read_offset;
 131
 132                 if (cpy > srcbufcpylen)
 133                         cpy = srcbufcpylen;
 134
 135                 if (userbuf) {
 136                         int notcopied = copy_to_user(dst, srcbufcpystart, cpy);
 137                         cpy -= notcopied;
 138                         if (unlikely(notcopied > 0))
 139                                 rc = -EFAULT;
 140                 } else {
 141                         memcpy(dst, srcbufcpystart, cpy);
 142                 }
 143
 144                 dst += cpy;
 145                 len -= cpy;
 146                 totalcpy += cpy;
 147
 148                 cn_l->data_buf.read_remaining -= cpy;
 149                 cn_l->data_buf.last_read_offset += cpy;
 150
 151                 if (cpy == srcbufcpylen)
 152                         databuf_nextreadchunk(cn_l);
 153
 154                 if (unlikely(rc < 0)) {
 155                         if (totalcpy == 0)
 156                                 totalcpy = rc;
 157                         break;
 158                 }
 159         }
 160
 161         return totalcpy;
 162 }
 163
 164 void databuf_pull(struct conn *cn_l, char *dst, int len)
 165 {
 166         _databuf_pull(cn_l, dst, len, 0);
 167 }
 168
 169 size_t databuf_pulluser(struct conn *trgt_sock_l, struct msghdr *msg)
 170 {
 171         size_t copied = 0;
 172         int iovidx = 0;
 173         int iovread = 0;
 174
 175         while (iovidx < msg->msg_iovlen) {
 176                 int rc;
 177
 178                 struct iovec *iov = msg->msg_iov + iovidx;
 179                 __user char *msg = iov->iov_base + iovread;
 180                 unsigned int len = iov->iov_len - iovread;
 181
 182                 if (len == 0) {
 183                         iovidx++;
 184                         iovread = 0;
 185                         continue;
 186                 }
 187
 188                 if (trgt_sock_l->data_buf.read_remaining == 0) {
 189                         rc = -EAGAIN;
 190                 } else {
 191                         if (len > trgt_sock_l->data_buf.read_remaining)
 192                                 len = trgt_sock_l->data_buf.read_remaining;
 193
 194                         rc = _databuf_pull(trgt_sock_l, msg, len, 1);
 195                 }
 196
 197                 BUG_ON(rc == 0);
 198
 199                 if (rc < 0) {
 200                         if (copied == 0)
 201                                 copied = rc;
 202                         break;
 203                 }
 204
 205                 copied += rc;
 206                 iovread += rc;
 207         }
 208
 209         return copied;
 210 }
 211
 212 void databuf_unpull(struct conn *trgt_out_l, __u32 bytes)
 213 {
 214         trgt_out_l->data_buf.read_remaining += bytes;
 215
 216         BUG_ON(trgt_out_l->data_buf.lastread == 0);
 217
 218         while (bytes > trgt_out_l->data_buf.last_read_offset) {
 219                 bytes -= trgt_out_l->data_buf.last_read_offset;
 220                 trgt_out_l->data_buf.lastread = container_of(
 221                                 trgt_out_l->data_buf.lastread->buf_list.prev,
 222                                 struct data_buf_item, buf_list);
 223                 BUG_ON(&(trgt_out_l->data_buf.lastread->buf_list) ==
 224                                 &(trgt_out_l->data_buf.items));
 225         }
 226
 227         trgt_out_l->data_buf.last_read_offset -= bytes;
 228 }
 229
 230 void databuf_pullold(struct conn *trgt_out_l, __u32 startpos, char *dst,
 231                 int len)
 232 {
 233         __u32 pos = trgt_out_l->data_buf.first_offset;
 234         struct data_buf_item *dbi = container_of(
 235                         trgt_out_l->data_buf.items.next,
 236                         struct data_buf_item, buf_list);
 237
 238         while(1) {
 239                 BUG_ON(&(dbi->buf_list) == &(trgt_out_l->data_buf.items));
 240
 241                 if (((__s32) (pos + dbi->datalen - startpos)) > 0)
 242                         break;
 243
 244                 pos += dbi->datalen;
 245                 dbi = container_of(dbi->buf_list.next, struct data_buf_item,
 246                                 buf_list);
 247         }
 248
 249         while (len > 0) {
 250                 int cpy = len;
 251
 252                 char *srcbufcpystart = 0;
 253                 int srcbufcpylen = 0;
 254
 255                 BUG_ON(&(dbi->buf_list) == &(trgt_out_l->data_buf.items));
 256
 257                 BUG_ON(((__s32) (pos - startpos)) > 0);
 258
 259                 srcbufcpystart = dbi->buf + ((__s32) (startpos - pos));
 260                 srcbufcpylen = dbi->datalen - ((__s32) (startpos - pos));
 261
 262                 if (cpy > srcbufcpylen)
 263                         cpy = srcbufcpylen;
 264
 265                 memcpy(dst, srcbufcpystart, cpy);
 266
 267                 dst += cpy;
 268                 len -= cpy;
 269                 startpos += cpy;
 270
 271                 pos += dbi->datalen;
 272                 dbi = container_of(dbi->buf_list.next, struct data_buf_item,
 273                                 buf_list);
 274         }
 275 }
 276
 277 /* ack up to *not* including pos */
 278 void databuf_ack(struct conn *trgt_out_l, __u32 pos)
 279 {
 280         __u32 acked = 0;
 281
 282         while (!list_empty(&(trgt_out_l->data_buf.items))) {
 283                 struct data_buf_item *firstitem = container_of(
 284                                 trgt_out_l->data_buf.items.next,
 285                                 struct data_buf_item, buf_list);
 286
 287                 if (firstitem == trgt_out_l->data_buf.lastread)
 288                         break;
 289
 290                 if ( ((__s32) (trgt_out_l->data_buf.first_offset +
 291                                 firstitem->datalen - pos)) > 0)
 292                         break;
 293
 294                 trgt_out_l->data_buf.first_offset += firstitem->datalen;
 295                 acked += firstitem->datalen;
 296
 297                 databuf_item_free(trgt_out_l, firstitem);
 298         }
 299
 300         trgt_out_l->data_buf.totalsize -= acked;
 301
 302         BUG_ON(trgt_out_l->data_buf.totalsize == 0 &&
 303                         trgt_out_l->data_buf.overhead != 0);
 304
 305         if (unlikely(trgt_out_l->data_buf.cpacket_buffer != 0)) {
 306                 __u32 amount = acked > trgt_out_l->data_buf.cpacket_buffer ?
 307                                 acked : trgt_out_l->data_buf.cpacket_buffer;
 308                 free_cpacket_buffer(amount);
 309                 trgt_out_l->data_buf.cpacket_buffer -= amount;
 310         }
 311
 312         if (trgt_out_l->sourcetype == SOURCE_IN)
 313                 refresh_speedstat(trgt_out_l, acked);
 314 }
 315
 316 void databuf_ackread(struct conn *cn_l)
 317 {
 318         __u32 acked = 0;
 319
 320         while (!list_empty(&(cn_l->data_buf.items)) &&
 321                         cn_l->data_buf.lastread != 0) {
 322                 struct data_buf_item *firstitem = container_of(
 323                                 cn_l->data_buf.items.next,
 324                                 struct data_buf_item, buf_list);
 325
 326                 if (firstitem == cn_l->data_buf.lastread)
 327                         break;
 328
 329                 acked += firstitem->datalen;
 330
 331                 databuf_item_free(cn_l, firstitem);
 332         }
 333
 334         cn_l->data_buf.first_offset += acked;
 335         cn_l->data_buf.totalsize -= acked;
 336
 337         BUG_ON(cn_l->data_buf.totalsize == 0 && cn_l->data_buf.overhead != 0);
 338
 339         if (unlikely(cn_l->data_buf.cpacket_buffer != 0)) {
 340                 __u32 amount = acked > cn_l->data_buf.cpacket_buffer ?
 341                                 acked : cn_l->data_buf.cpacket_buffer;
 342                 free_cpacket_buffer(amount);
 343                 cn_l->data_buf.cpacket_buffer -= amount;
 344         }
 345
 346         if (cn_l->sourcetype == SOURCE_IN)
 347                 refresh_speedstat(cn_l, acked);
 348 }
 349
 350 static __s64 _receive_buf(struct conn *cn_l, char *buf, __u32 len, int userbuf,
 351                 __u32 maxcpy, __u32 maxusage)
 352 {
 353         struct data_buf_item *item = 0;
 354
 355         __s64 totalcpy = 0;
 356
 357         if (list_empty(&(cn_l->data_buf.items)) == 0) {
 358                 struct list_head *last = cn_l->data_buf.items.prev;
 359                 item = container_of(last, struct data_buf_item, buf_list);
 360         }
 361
 362         while (len > 0) {
 363                 int rc = 0;
 364                 int cpy = len;
 365
 366                 if (item == 0 || item->buflen <= item->datalen) {
 367                         __u32 buflen = len;
 368
 369                         if (maxusage != 0) {
 370                                 if (cn_l->data_buf.totalsize +
 371                                                 cn_l->data_buf.overhead >
 372                                                 maxusage) {
 373                                         rc = -EAGAIN;
 374                                         goto error;
 375                                 }
 376
 377                                 buflen = maxusage - cn_l->data_buf.totalsize -
 378                                                 cn_l->data_buf.overhead -
 379                                                 sizeof(struct data_buf_item);
 380                         } else {
 381                                 if (totalcpy + 64 > maxcpy &&
 382                                                 totalcpy + len > maxcpy) {
 383                                         rc = -EAGAIN;
 384                                         goto error;
 385                                 }
 386
 387                                 if (totalcpy + buflen < maxcpy)
 388                                         buflen = maxcpy - totalcpy;
 389                         }
 390
 391                         if (buflen < 64)
 392                                 buflen = 64;
 393                         if (buflen > PAGESIZE)
 394                                 buflen = PAGESIZE;
 395                         if (buflen > 32768)
 396                                 buflen = 32768;
 397
 398                         if (unlikely(unlikely(cn_l->data_buf.totalsize +
 399                                         buflen > (1 << 30)) || unlikely(
 400                                         cn_l->data_buf.overhead > (1 << 30)))) {
 401                                 rc = -EAGAIN;
 402                                 goto error;
 403                         }
 404
 405                         item = kmem_cache_alloc(data_buf_item_slab, GFP_KERNEL);
 406                         if (unlikely(item == 0)) {
 407                                 rc = -ENOMEM;
 408                                 goto error;
 409                         }
 410                         memset(item, 0, sizeof(item));
 411                         item->type = TYPE_BUF;
 412                         item->buf = kmalloc(buflen, GFP_KERNEL);
 413
 414                         if (unlikely(item->buf == 0)) {
 415                                 kmem_cache_free(data_buf_item_slab, item);
 416                                 rc = -ENOMEM;
 417                                 goto error;
 418                         }
 419                         item->datalen = 0;
 420
 421                         list_add_tail(&(item->buf_list),
 422                                         &(cn_l->data_buf.items));
 423                         item->buflen = buflen;
 424                         cn_l->data_buf.overhead += buflen +
 425                                         sizeof(struct data_buf_item);
 426                 }
 427
 428                 BUG_ON(item->type != TYPE_BUF);
 429                 BUG_ON(item->buflen <= item->datalen);
 430
 431                 if (item->buflen - item->datalen < cpy)
 432                         cpy = (item->buflen - item->datalen);
 433
 434                 if (userbuf) {
 435                         int notcopied = copy_from_user(item->buf +
 436                                         item->datalen, buf, cpy);
 437                         cpy -= notcopied;
 438                         if (unlikely(notcopied > 0))
 439                                 rc = -EFAULT;
 440                 } else {
 441                         memcpy(item->buf + item->datalen, buf, cpy);
 442                 }
 443
 444                 buf += cpy;
 445                 len -= cpy;
 446                 cn_l->data_buf.read_remaining += cpy;
 447                 cn_l->data_buf.totalsize += cpy;
 448                 cn_l->data_buf.overhead -= cpy;
 449                 BUG_ON(cn_l->data_buf.totalsize == 0 &&
 450                                 cn_l->data_buf.overhead != 0);
 451                 totalcpy += cpy;
 452
 453                 item->datalen += cpy;
 454
 455 error:
 456                 if (unlikely(rc < 0)) {
 457                         if (totalcpy == 0)
 458                                 return rc;
 459                         break;
 460                 }
 461         }
 462
 463         return totalcpy;
 464 }
 465
 466 __s64 receive_userbuf(struct conn *src_sock_l, struct msghdr *msg, __u32 maxcpy,
 467                 __u32 maxusage)
 468 {
 469         __s64 copied = 0;
 470         int iovidx = 0;
 471         int iovread = 0;
 472
 473         while (iovidx < msg->msg_iovlen) {
 474                 struct iovec *iov = msg->msg_iov + iovidx;
 475                 __user char *userbuf = iov->iov_base + iovread;
 476                 __u32 len = iov->iov_len - iovread;
 477                 __s64 rc;
 478
 479                 if (len == 0) {
 480                         iovidx++;
 481                         iovread = 0;
 482                         continue;
 483                 }
 484
 485                 BUG_ON(copied < 0);
 486                 BUG_ON(copied > maxcpy);
 487                 rc = _receive_buf(src_sock_l, userbuf, len, 1, maxcpy - copied,
 488                                 maxusage);
 489
 490                 if (rc < 0) {
 491                         if (copied == 0)
 492                                 copied = rc;
 493                         break;
 494                 }
 495
 496                 copied += rc;
 497                 iovread += rc;
 498
 499                 if (rc < len)
 500                         break;
 501         }
 502
 503         return copied;
 504 }
 505
 506 void receive_cpacketresp(struct conn *trtg_unconn_l, char *buf, int len)
 507 {
 508         __s64 rc;
 509         BUG_ON(trtg_unconn_l->data_buf.cpacket_buffer <
 510                         trtg_unconn_l->data_buf.totalsize + len);
 511         rc = _receive_buf(trtg_unconn_l, buf, len, 0, len, 0);
 512         BUG_ON(rc < 0);
 513         BUG_ON(rc < len);
 514 }
 515
 516 int receive_skb(struct conn *src_in_l, struct sk_buff *skb)
 517 {
 518         struct skb_procstate *ps = skb_pstate(skb);
 519         struct data_buf_item *item = &(ps->funcstate.rcv2.dbi);
 520
 521         if (unlikely(unlikely(src_in_l->data_buf.totalsize + skb->len >
 522                         (1 << 30)) || unlikely(src_in_l->data_buf.overhead >
 523                         (1 << 30))))
 524                 return 1;
 525
 526         item->type = TYPE_SKB;
 527         item->buf = skb->data;
 528         item->datalen = skb->len;
 529         item->buflen = item->datalen;
 530         list_add_tail(&(item->buf_list), &(src_in_l->data_buf.items));
 531
 532         src_in_l->data_buf.read_remaining += item->datalen;
 533         src_in_l->data_buf.totalsize += item->datalen;
 534         src_in_l->data_buf.overhead += sizeof(struct sk_buff);
 535
 536         return 0;
 537 }
 538
 539 static void _wake_sender_in(struct conn *src_in_l)
 540 {
 541         drain_ooo_queue(src_in_l);
 542         mutex_unlock(&(src_in_l->rcv_lock));
 543         get_window(src_in_l, 0, 0, 0);
 544 }
 545
 546 void wake_sender(struct conn *cn)
 547 {
 548         unreserve_sock_buffer(cn);
 549
 550         mutex_lock(&(cn->rcv_lock));
 551         switch (cn->sourcetype) {
 552         case SOURCE_NONE:
 553                 mutex_unlock(&(cn->rcv_lock));
 554                 parse(cn->reversedir, 0);
 555                 break;
 556         case SOURCE_SOCK:
 557                 wake_up_interruptible(&(cn->source.sock.wait));
 558                 mutex_unlock(&(cn->rcv_lock));
 559                 break;
 560         case SOURCE_IN:
 561                 _wake_sender_in(cn); /* mutex_unlock inside */
 562                 break;
 563         default:
 564                 BUG();
 565         }
 566 }
 567
 568 void flush_buf(struct conn *cn)
 569 {
 570         int rc;
 571         int sent = 0;
 572         mutex_lock(&(cn->rcv_lock));
 573
 574         switch (cn->targettype) {
 575         case TARGET_UNCONNECTED:
 576                 mutex_unlock(&(cn->rcv_lock));
 577                 parse(cn, 0);
 578                 break;
 579         case TARGET_SOCK:
 580                 if (cn->sourcetype != SOURCE_SOCK ||
 581                                 cn->source.sock.delay_flush == 0 ||
 582                                 cn->data_buf.totalsize +
 583                                 cn->data_buf.overhead -
 584                                 cn->data_buf.cpacket_buffer >=
 585                                 BUFFERLIMIT_SOCK_SOCK/2)
 586                         wake_up_interruptible(&(cn->target.sock.wait));
 587                 mutex_unlock(&(cn->rcv_lock));
 588                 break;
 589         case TARGET_OUT:
 590                 rc = flush_out(cn, 0, 0);
 591                 mutex_unlock(&(cn->rcv_lock));
 592                 sent = (rc == RC_FLUSH_CONN_OUT_OK_SENT);
 593                 break;
 594         case TARGET_DISCARD:
 595                 databuf_ackdiscard(cn);
 596                 mutex_unlock(&(cn->rcv_lock));
 597                 sent = 1;
 598                 break;
 599         default:
 600                 BUG();
 601         }
 602
 603         refresh_conn_credits(cn, 0, 0);
 604
 605         if (sent) {
 606                 wake_sender(cn);
 607         }
 608 }
 609
 610 void __init forward_init(void)
 611 {
 612         data_buf_item_slab = kmem_cache_create("cor_data_buf_item",
 613                         sizeof(struct data_buf_item), 8, 0, 0);
 614 }
 615
 616 MODULE_LICENSE("GPL");