net/rxrpc/call.c

   1 /* call.c: Rx call routines
   2  *
   3  * Copyright (C) 2002 Red Hat, Inc. All Rights Reserved.
   4  * Written by David Howells (dhowells@redhat.com)
   5  *
   6  * This program is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU General Public License
   8  * as published by the Free Software Foundation; either version
   9  * 2 of the License, or (at your option) any later version.
  10  */
  11
  12 #include <linux/sched.h>
  13 #include <linux/slab.h>
  14 #include <linux/module.h>
  15 #include <rxrpc/rxrpc.h>
  16 #include <rxrpc/transport.h>
  17 #include <rxrpc/peer.h>
  18 #include <rxrpc/connection.h>
  19 #include <rxrpc/call.h>
  20 #include <rxrpc/message.h>
  21 #include "internal.h"
  22
  23 __RXACCT_DECL(atomic_t rxrpc_call_count);
  24 __RXACCT_DECL(atomic_t rxrpc_message_count);
  25
  26 LIST_HEAD(rxrpc_calls);
  27 DECLARE_RWSEM(rxrpc_calls_sem);
  28
  29 unsigned rxrpc_call_rcv_timeout         = HZ/3;
  30 unsigned rxrpc_call_acks_timeout        = HZ/3;
  31 unsigned rxrpc_call_dfr_ack_timeout     = HZ/20;
  32 unsigned short rxrpc_call_max_resend    = HZ/10;
  33
  34 const char *rxrpc_call_states[] = {
  35         "COMPLETE",
  36         "ERROR",
  37         "SRVR_RCV_OPID",
  38         "SRVR_RCV_ARGS",
  39         "SRVR_GOT_ARGS",
  40         "SRVR_SND_REPLY",
  41         "SRVR_RCV_FINAL_ACK",
  42         "CLNT_SND_ARGS",
  43         "CLNT_RCV_REPLY",
  44         "CLNT_GOT_REPLY"
  45 };
  46
  47 const char *rxrpc_call_error_states[] = {
  48         "NO_ERROR",
  49         "LOCAL_ABORT",
  50         "PEER_ABORT",
  51         "LOCAL_ERROR",
  52         "REMOTE_ERROR"
  53 };
  54
  55 const char *rxrpc_pkts[] = {
  56         "?00",
  57         "data", "ack", "busy", "abort", "ackall", "chall", "resp", "debug",
  58         "?09", "?10", "?11", "?12", "?13", "?14", "?15"
  59 };
  60
  61 const char *rxrpc_acks[] = {
  62         "---", "REQ", "DUP", "SEQ", "WIN", "MEM", "PNG", "PNR", "DLY", "IDL",
  63         "-?-"
  64 };
  65
  66 static const char _acktype[] = "NA-";
  67
  68 static void rxrpc_call_receive_packet(struct rxrpc_call *call);
  69 static void rxrpc_call_receive_data_packet(struct rxrpc_call *call,
  70                                            struct rxrpc_message *msg);
  71 static void rxrpc_call_receive_ack_packet(struct rxrpc_call *call,
  72                                           struct rxrpc_message *msg);
  73 static void rxrpc_call_definitively_ACK(struct rxrpc_call *call,
  74                                         rxrpc_seq_t higest);
  75 static void rxrpc_call_resend(struct rxrpc_call *call, rxrpc_seq_t highest);
  76 static int __rxrpc_call_read_data(struct rxrpc_call *call);
  77
  78 static int rxrpc_call_record_ACK(struct rxrpc_call *call,
  79                                  struct rxrpc_message *msg,
  80                                  rxrpc_seq_t seq,
  81                                  size_t count);
  82 #define _state(call) \
  83         _debug("[[[ state %s ]]]", rxrpc_call_states[call->app_call_state]);
  84
  85 static void rxrpc_call_default_attn_func(struct rxrpc_call *call)
  86 {
  87         wake_up(&call->waitq);
  88 }
  89
  90 static void rxrpc_call_default_error_func(struct rxrpc_call *call)
  91 {
  92         wake_up(&call->waitq);
  93 }
  94
  95 static void rxrpc_call_default_aemap_func(struct rxrpc_call *call)
  96 {
  97         switch (call->app_err_state) {
  98         case RXRPC_ESTATE_LOCAL_ABORT:
  99                 call->app_abort_code = -call->app_errno;
 100         case RXRPC_ESTATE_PEER_ABORT:
 101                 call->app_errno = -ECONNABORTED;
 102         default:
 103                 break;
 104         }
 105 }
 106
 107 static void __rxrpc_call_acks_timeout(unsigned long _call)
 108 {
 109         struct rxrpc_call *call = (struct rxrpc_call *) _call;
 110
 111         _debug("ACKS TIMEOUT %05lu", jiffies - call->cjif);
 112
 113         call->flags |= RXRPC_CALL_ACKS_TIMO;
 114         rxrpc_krxiod_queue_call(call);
 115 }
 116
 117 static void __rxrpc_call_rcv_timeout(unsigned long _call)
 118 {
 119         struct rxrpc_call *call = (struct rxrpc_call *) _call;
 120
 121         _debug("RCV TIMEOUT %05lu", jiffies - call->cjif);
 122
 123         call->flags |= RXRPC_CALL_RCV_TIMO;
 124         rxrpc_krxiod_queue_call(call);
 125 }
 126
 127 static void __rxrpc_call_ackr_timeout(unsigned long _call)
 128 {
 129         struct rxrpc_call *call = (struct rxrpc_call *) _call;
 130
 131         _debug("ACKR TIMEOUT %05lu",jiffies - call->cjif);
 132
 133         call->flags |= RXRPC_CALL_ACKR_TIMO;
 134         rxrpc_krxiod_queue_call(call);
 135 }
 136
 137 /*****************************************************************************/
 138 /*
 139  * calculate a timeout based on an RTT value
 140  */
 141 static inline unsigned long __rxrpc_rtt_based_timeout(struct rxrpc_call *call,
 142                                                       unsigned long val)
 143 {
 144         unsigned long expiry = call->conn->peer->rtt / (1000000 / HZ);
 145
 146         expiry += 10;
 147         if (expiry < HZ / 25)
 148                 expiry = HZ / 25;
 149         if (expiry > HZ)
 150                 expiry = HZ;
 151
 152         _leave(" = %lu jiffies", expiry);
 153         return jiffies + expiry;
 154 } /* end __rxrpc_rtt_based_timeout() */
 155
 156 /*****************************************************************************/
 157 /*
 158  * create a new call record
 159  */
 160 static inline int __rxrpc_create_call(struct rxrpc_connection *conn,
 161                                       struct rxrpc_call **_call)
 162 {
 163         struct rxrpc_call *call;
 164
 165         _enter("%p", conn);
 166
 167         /* allocate and initialise a call record */
 168         call = (struct rxrpc_call *) get_zeroed_page(GFP_KERNEL);
 169         if (!call) {
 170                 _leave(" ENOMEM");
 171                 return -ENOMEM;
 172         }
 173
 174         atomic_set(&call->usage, 1);
 175
 176         init_waitqueue_head(&call->waitq);
 177         spin_lock_init(&call->lock);
 178         INIT_LIST_HEAD(&call->link);
 179         INIT_LIST_HEAD(&call->acks_pendq);
 180         INIT_LIST_HEAD(&call->rcv_receiveq);
 181         INIT_LIST_HEAD(&call->rcv_krxiodq_lk);
 182         INIT_LIST_HEAD(&call->app_readyq);
 183         INIT_LIST_HEAD(&call->app_unreadyq);
 184         INIT_LIST_HEAD(&call->app_link);
 185         INIT_LIST_HEAD(&call->app_attn_link);
 186
 187         init_timer(&call->acks_timeout);
 188         call->acks_timeout.data = (unsigned long) call;
 189         call->acks_timeout.function = __rxrpc_call_acks_timeout;
 190
 191         init_timer(&call->rcv_timeout);
 192         call->rcv_timeout.data = (unsigned long) call;
 193         call->rcv_timeout.function = __rxrpc_call_rcv_timeout;
 194
 195         init_timer(&call->ackr_dfr_timo);
 196         call->ackr_dfr_timo.data = (unsigned long) call;
 197         call->ackr_dfr_timo.function = __rxrpc_call_ackr_timeout;
 198
 199         call->conn = conn;
 200         call->ackr_win_bot = 1;
 201         call->ackr_win_top = call->ackr_win_bot + RXRPC_CALL_ACK_WINDOW_SIZE - 1;
 202         call->ackr_prev_seq = 0;
 203         call->app_mark = RXRPC_APP_MARK_EOF;
 204         call->app_attn_func = rxrpc_call_default_attn_func;
 205         call->app_error_func = rxrpc_call_default_error_func;
 206         call->app_aemap_func = rxrpc_call_default_aemap_func;
 207         call->app_scr_alloc = call->app_scratch;
 208
 209         call->cjif = jiffies;
 210
 211         _leave(" = 0 (%p)", call);
 212
 213         *_call = call;
 214
 215         return 0;
 216 } /* end __rxrpc_create_call() */
 217
 218 /*****************************************************************************/
 219 /*
 220  * create a new call record for outgoing calls
 221  */
 222 int rxrpc_create_call(struct rxrpc_connection *conn,
 223                       rxrpc_call_attn_func_t attn,
 224                       rxrpc_call_error_func_t error,
 225                       rxrpc_call_aemap_func_t aemap,
 226                       struct rxrpc_call **_call)
 227 {
 228         DECLARE_WAITQUEUE(myself, current);
 229
 230         struct rxrpc_call *call;
 231         int ret, cix, loop;
 232
 233         _enter("%p", conn);
 234
 235         /* allocate and initialise a call record */
 236         ret = __rxrpc_create_call(conn, &call);
 237         if (ret < 0) {
 238                 _leave(" = %d", ret);
 239                 return ret;
 240         }
 241
 242         call->app_call_state = RXRPC_CSTATE_CLNT_SND_ARGS;
 243         if (attn)
 244                 call->app_attn_func = attn;
 245         if (error)
 246                 call->app_error_func = error;
 247         if (aemap)
 248                 call->app_aemap_func = aemap;
 249
 250         _state(call);
 251
 252         spin_lock(&conn->lock);
 253         set_current_state(TASK_INTERRUPTIBLE);
 254         add_wait_queue(&conn->chanwait, &myself);
 255
 256  try_again:
 257         /* try to find an unused channel */
 258         for (cix = 0; cix < 4; cix++)
 259                 if (!conn->channels[cix])
 260                         goto obtained_chan;
 261
 262         /* no free channels - wait for one to become available */
 263         ret = -EINTR;
 264         if (signal_pending(current))
 265                 goto error_unwait;
 266
 267         spin_unlock(&conn->lock);
 268
 269         schedule();
 270         set_current_state(TASK_INTERRUPTIBLE);
 271
 272         spin_lock(&conn->lock);
 273         goto try_again;
 274
 275         /* got a channel - now attach to the connection */
 276  obtained_chan:
 277         remove_wait_queue(&conn->chanwait, &myself);
 278         set_current_state(TASK_RUNNING);
 279
 280         /* concoct a unique call number */
 281  next_callid:
 282         call->call_id = htonl(++conn->call_counter);
 283         for (loop = 0; loop < 4; loop++)
 284                 if (conn->channels[loop] &&
 285                     conn->channels[loop]->call_id == call->call_id)
 286                         goto next_callid;
 287
 288         rxrpc_get_connection(conn);
 289         conn->channels[cix] = call; /* assign _after_ done callid check loop */
 290         do_gettimeofday(&conn->atime);
 291         call->chan_ix = htonl(cix);
 292
 293         spin_unlock(&conn->lock);
 294
 295         down_write(&rxrpc_calls_sem);
 296         list_add_tail(&call->call_link, &rxrpc_calls);
 297         up_write(&rxrpc_calls_sem);
 298
 299         __RXACCT(atomic_inc(&rxrpc_call_count));
 300         *_call = call;
 301
 302         _leave(" = 0 (call=%p cix=%u)", call, cix);
 303         return 0;
 304
 305  error_unwait:
 306         remove_wait_queue(&conn->chanwait, &myself);
 307         set_current_state(TASK_RUNNING);
 308         spin_unlock(&conn->lock);
 309
 310         free_page((unsigned long) call);
 311         _leave(" = %d", ret);
 312         return ret;
 313 } /* end rxrpc_create_call() */
 314
 315 /*****************************************************************************/
 316 /*
 317  * create a new call record for incoming calls
 318  */
 319 int rxrpc_incoming_call(struct rxrpc_connection *conn,
 320                         struct rxrpc_message *msg,
 321                         struct rxrpc_call **_call)
 322 {
 323         struct rxrpc_call *call;
 324         unsigned cix;
 325         int ret;
 326
 327         cix = ntohl(msg->hdr.cid) & RXRPC_CHANNELMASK;
 328
 329         _enter("%p,%u,%u", conn, ntohl(msg->hdr.callNumber), cix);
 330
 331         /* allocate and initialise a call record */
 332         ret = __rxrpc_create_call(conn, &call);
 333         if (ret < 0) {
 334                 _leave(" = %d", ret);
 335                 return ret;
 336         }
 337
 338         call->pkt_rcv_count = 1;
 339         call->app_call_state = RXRPC_CSTATE_SRVR_RCV_OPID;
 340         call->app_mark = sizeof(uint32_t);
 341
 342         _state(call);
 343
 344         /* attach to the connection */
 345         ret = -EBUSY;
 346         call->chan_ix = htonl(cix);
 347         call->call_id = msg->hdr.callNumber;
 348
 349         spin_lock(&conn->lock);
 350
 351         if (!conn->channels[cix] ||
 352             conn->channels[cix]->app_call_state == RXRPC_CSTATE_COMPLETE ||
 353             conn->channels[cix]->app_call_state == RXRPC_CSTATE_ERROR
 354             ) {
 355                 conn->channels[cix] = call;
 356                 rxrpc_get_connection(conn);
 357                 ret = 0;
 358         }
 359
 360         spin_unlock(&conn->lock);
 361
 362         if (ret < 0) {
 363                 free_page((unsigned long) call);
 364                 call = NULL;
 365         }
 366
 367         if (ret == 0) {
 368                 down_write(&rxrpc_calls_sem);
 369                 list_add_tail(&call->call_link, &rxrpc_calls);
 370                 up_write(&rxrpc_calls_sem);
 371                 __RXACCT(atomic_inc(&rxrpc_call_count));
 372                 *_call = call;
 373         }
 374
 375         _leave(" = %d [%p]", ret, call);
 376         return ret;
 377 } /* end rxrpc_incoming_call() */
 378
 379 /*****************************************************************************/
 380 /*
 381  * free a call record
 382  */
 383 void rxrpc_put_call(struct rxrpc_call *call)
 384 {
 385         struct rxrpc_connection *conn = call->conn;
 386         struct rxrpc_message *msg;
 387
 388         _enter("%p{u=%d}",call,atomic_read(&call->usage));
 389
 390         /* sanity check */
 391         if (atomic_read(&call->usage) <= 0)
 392                 BUG();
 393
 394         /* to prevent a race, the decrement and the de-list must be effectively
 395          * atomic */
 396         spin_lock(&conn->lock);
 397         if (likely(!atomic_dec_and_test(&call->usage))) {
 398                 spin_unlock(&conn->lock);
 399                 _leave("");
 400                 return;
 401         }
 402
 403         if (conn->channels[ntohl(call->chan_ix)] == call)
 404                 conn->channels[ntohl(call->chan_ix)] = NULL;
 405
 406         spin_unlock(&conn->lock);
 407
 408         wake_up(&conn->chanwait);
 409
 410         rxrpc_put_connection(conn);
 411
 412         /* clear the timers and dequeue from krxiod */
 413         del_timer_sync(&call->acks_timeout);
 414         del_timer_sync(&call->rcv_timeout);
 415         del_timer_sync(&call->ackr_dfr_timo);
 416
 417         rxrpc_krxiod_dequeue_call(call);
 418
 419         /* clean up the contents of the struct */
 420         if (call->snd_nextmsg)
 421                 rxrpc_put_message(call->snd_nextmsg);
 422
 423         if (call->snd_ping)
 424                 rxrpc_put_message(call->snd_ping);
 425
 426         while (!list_empty(&call->acks_pendq)) {
 427                 msg = list_entry(call->acks_pendq.next,
 428                                  struct rxrpc_message, link);
 429                 list_del(&msg->link);
 430                 rxrpc_put_message(msg);
 431         }
 432
 433         while (!list_empty(&call->rcv_receiveq)) {
 434                 msg = list_entry(call->rcv_receiveq.next,
 435                                  struct rxrpc_message, link);
 436                 list_del(&msg->link);
 437                 rxrpc_put_message(msg);
 438         }
 439
 440         while (!list_empty(&call->app_readyq)) {
 441                 msg = list_entry(call->app_readyq.next,
 442                                  struct rxrpc_message, link);
 443                 list_del(&msg->link);
 444                 rxrpc_put_message(msg);
 445         }
 446
 447         while (!list_empty(&call->app_unreadyq)) {
 448                 msg = list_entry(call->app_unreadyq.next,
 449                                  struct rxrpc_message, link);
 450                 list_del(&msg->link);
 451                 rxrpc_put_message(msg);
 452         }
 453
 454         module_put(call->owner);
 455
 456         down_write(&rxrpc_calls_sem);
 457         list_del(&call->call_link);
 458         up_write(&rxrpc_calls_sem);
 459
 460         __RXACCT(atomic_dec(&rxrpc_call_count));
 461         free_page((unsigned long) call);
 462
 463         _leave(" [destroyed]");
 464 } /* end rxrpc_put_call() */
 465
 466 /*****************************************************************************/
 467 /*
 468  * actually generate a normal ACK
 469  */
 470 static inline int __rxrpc_call_gen_normal_ACK(struct rxrpc_call *call,
 471                                               rxrpc_seq_t seq)
 472 {
 473         struct rxrpc_message *msg;
 474         struct kvec diov[3];
 475         __be32 aux[4];
 476         int delta, ret;
 477
 478         /* ACKs default to DELAY */
 479         if (!call->ackr.reason)
 480                 call->ackr.reason = RXRPC_ACK_DELAY;
 481
 482         _proto("Rx %05lu Sending ACK { m=%hu f=#%u p=#%u s=%%%u r=%s n=%u }",
 483                jiffies - call->cjif,
 484                ntohs(call->ackr.maxSkew),
 485                ntohl(call->ackr.firstPacket),
 486                ntohl(call->ackr.previousPacket),
 487                ntohl(call->ackr.serial),
 488                rxrpc_acks[call->ackr.reason],
 489                call->ackr.nAcks);
 490
 491         aux[0] = htonl(call->conn->peer->if_mtu);       /* interface MTU */
 492         aux[1] = htonl(1444);                           /* max MTU */
 493         aux[2] = htonl(16);                             /* rwind */
 494         aux[3] = htonl(4);                              /* max packets */
 495
 496         diov[0].iov_len  = sizeof(struct rxrpc_ackpacket);
 497         diov[0].iov_base = &call->ackr;
 498         diov[1].iov_len  = call->ackr_pend_cnt + 3;
 499         diov[1].iov_base = call->ackr_array;
 500         diov[2].iov_len  = sizeof(aux);
 501         diov[2].iov_base = &aux;
 502
 503         /* build and send the message */
 504         ret = rxrpc_conn_newmsg(call->conn,call, RXRPC_PACKET_TYPE_ACK,
 505                                 3, diov, GFP_KERNEL, &msg);
 506         if (ret < 0)
 507                 goto out;
 508
 509         msg->seq = seq;
 510         msg->hdr.seq = htonl(seq);
 511         msg->hdr.flags |= RXRPC_SLOW_START_OK;
 512
 513         ret = rxrpc_conn_sendmsg(call->conn, msg);
 514         rxrpc_put_message(msg);
 515         if (ret < 0)
 516                 goto out;
 517         call->pkt_snd_count++;
 518
 519         /* count how many actual ACKs there were at the front */
 520         for (delta = 0; delta < call->ackr_pend_cnt; delta++)
 521                 if (call->ackr_array[delta] != RXRPC_ACK_TYPE_ACK)
 522                         break;
 523
 524         call->ackr_pend_cnt -= delta; /* all ACK'd to this point */
 525
 526         /* crank the ACK window around */
 527         if (delta == 0) {
 528                 /* un-ACK'd window */
 529         }
 530         else if (delta < RXRPC_CALL_ACK_WINDOW_SIZE) {
 531                 /* partially ACK'd window
 532                  * - shuffle down to avoid losing out-of-sequence packets
 533                  */
 534                 call->ackr_win_bot += delta;
 535                 call->ackr_win_top += delta;
 536
 537                 memmove(&call->ackr_array[0],
 538                         &call->ackr_array[delta],
 539                         call->ackr_pend_cnt);
 540
 541                 memset(&call->ackr_array[call->ackr_pend_cnt],
 542                        RXRPC_ACK_TYPE_NACK,
 543                        sizeof(call->ackr_array) - call->ackr_pend_cnt);
 544         }
 545         else {
 546                 /* fully ACK'd window
 547                  * - just clear the whole thing
 548                  */
 549                 memset(&call->ackr_array,
 550                        RXRPC_ACK_TYPE_NACK,
 551                        sizeof(call->ackr_array));
 552         }
 553
 554         /* clear this ACK */
 555         memset(&call->ackr, 0, sizeof(call->ackr));
 556
 557  out:
 558         if (!call->app_call_state)
 559                 printk("___ STATE 0 ___\n");
 560         return ret;
 561 } /* end __rxrpc_call_gen_normal_ACK() */
 562
 563 /*****************************************************************************/
 564 /*
 565  * note the reception of a packet in the call's ACK records and generate an
 566  * appropriate ACK packet if necessary
 567  * - returns 0 if packet should be processed, 1 if packet should be ignored
 568  *   and -ve on an error
 569  */
 570 static int rxrpc_call_generate_ACK(struct rxrpc_call *call,
 571                                    struct rxrpc_header *hdr,
 572                                    struct rxrpc_ackpacket *ack)
 573 {
 574         struct rxrpc_message *msg;
 575         rxrpc_seq_t seq;
 576         unsigned offset;
 577         int ret = 0, err;
 578         u8 special_ACK, do_ACK, force;
 579
 580         _enter("%p,%p { seq=%d tp=%d fl=%02x }",
 581                call, hdr, ntohl(hdr->seq), hdr->type, hdr->flags);
 582
 583         seq = ntohl(hdr->seq);
 584         offset = seq - call->ackr_win_bot;
 585         do_ACK = RXRPC_ACK_DELAY;
 586         special_ACK = 0;
 587         force = (seq == 1);
 588
 589         if (call->ackr_high_seq < seq)
 590                 call->ackr_high_seq = seq;
 591
 592         /* deal with generation of obvious special ACKs first */
 593         if (ack && ack->reason == RXRPC_ACK_PING) {
 594                 special_ACK = RXRPC_ACK_PING_RESPONSE;
 595                 ret = 1;
 596                 goto gen_ACK;
 597         }
 598
 599         if (seq < call->ackr_win_bot) {
 600                 special_ACK = RXRPC_ACK_DUPLICATE;
 601                 ret = 1;
 602                 goto gen_ACK;
 603         }
 604
 605         if (seq >= call->ackr_win_top) {
 606                 special_ACK = RXRPC_ACK_EXCEEDS_WINDOW;
 607                 ret = 1;
 608                 goto gen_ACK;
 609         }
 610
 611         if (call->ackr_array[offset] != RXRPC_ACK_TYPE_NACK) {
 612                 special_ACK = RXRPC_ACK_DUPLICATE;
 613                 ret = 1;
 614                 goto gen_ACK;
 615         }
 616
 617         /* okay... it's a normal data packet inside the ACK window */
 618         call->ackr_array[offset] = RXRPC_ACK_TYPE_ACK;
 619
 620         if (offset < call->ackr_pend_cnt) {
 621         }
 622         else if (offset > call->ackr_pend_cnt) {
 623                 do_ACK = RXRPC_ACK_OUT_OF_SEQUENCE;
 624                 call->ackr_pend_cnt = offset;
 625                 goto gen_ACK;
 626         }
 627
 628         if (hdr->flags & RXRPC_REQUEST_ACK) {
 629                 do_ACK = RXRPC_ACK_REQUESTED;
 630         }
 631
 632         /* generate an ACK on the final packet of a reply just received */
 633         if (hdr->flags & RXRPC_LAST_PACKET) {
 634                 if (call->conn->out_clientflag)
 635                         force = 1;
 636         }
 637         else if (!(hdr->flags & RXRPC_MORE_PACKETS)) {
 638                 do_ACK = RXRPC_ACK_REQUESTED;
 639         }
 640
 641         /* re-ACK packets previously received out-of-order */
 642         for (offset++; offset < RXRPC_CALL_ACK_WINDOW_SIZE; offset++)
 643                 if (call->ackr_array[offset] != RXRPC_ACK_TYPE_ACK)
 644                         break;
 645
 646         call->ackr_pend_cnt = offset;
 647
 648         /* generate an ACK if we fill up the window */
 649         if (call->ackr_pend_cnt >= RXRPC_CALL_ACK_WINDOW_SIZE)
 650                 force = 1;
 651
 652  gen_ACK:
 653         _debug("%05lu ACKs pend=%u norm=%s special=%s%s",
 654                jiffies - call->cjif,
 655                call->ackr_pend_cnt,
 656                rxrpc_acks[do_ACK],
 657                rxrpc_acks[special_ACK],
 658                force ? " immediate" :
 659                do_ACK == RXRPC_ACK_REQUESTED ? " merge-req" :
 660                hdr->flags & RXRPC_LAST_PACKET ? " finalise" :
 661                " defer"
 662                );
 663
 664         /* send any pending normal ACKs if need be */
 665         if (call->ackr_pend_cnt > 0) {
 666                 /* fill out the appropriate form */
 667                 call->ackr.bufferSpace  = htons(RXRPC_CALL_ACK_WINDOW_SIZE);
 668                 call->ackr.maxSkew      = htons(min(call->ackr_high_seq - seq,
 669                                                     65535U));
 670                 call->ackr.firstPacket  = htonl(call->ackr_win_bot);
 671                 call->ackr.previousPacket = call->ackr_prev_seq;
 672                 call->ackr.serial       = hdr->serial;
 673                 call->ackr.nAcks        = call->ackr_pend_cnt;
 674
 675                 if (do_ACK == RXRPC_ACK_REQUESTED)
 676                         call->ackr.reason = do_ACK;
 677
 678                 /* generate the ACK immediately if necessary */
 679                 if (special_ACK || force) {
 680                         err = __rxrpc_call_gen_normal_ACK(
 681                                 call, do_ACK == RXRPC_ACK_DELAY ? 0 : seq);
 682                         if (err < 0) {
 683                                 ret = err;
 684                                 goto out;
 685                         }
 686                 }
 687         }
 688
 689         if (call->ackr.reason == RXRPC_ACK_REQUESTED)
 690                 call->ackr_dfr_seq = seq;
 691
 692         /* start the ACK timer if not running if there are any pending deferred
 693          * ACKs */
 694         if (call->ackr_pend_cnt > 0 &&
 695             call->ackr.reason != RXRPC_ACK_REQUESTED &&
 696             !timer_pending(&call->ackr_dfr_timo)
 697             ) {
 698                 unsigned long timo;
 699
 700                 timo = rxrpc_call_dfr_ack_timeout + jiffies;
 701
 702                 _debug("START ACKR TIMER for cj=%lu", timo - call->cjif);
 703
 704                 spin_lock(&call->lock);
 705                 mod_timer(&call->ackr_dfr_timo, timo);
 706                 spin_unlock(&call->lock);
 707         }
 708         else if ((call->ackr_pend_cnt == 0 ||
 709                   call->ackr.reason == RXRPC_ACK_REQUESTED) &&
 710                  timer_pending(&call->ackr_dfr_timo)
 711                  ) {
 712                 /* stop timer if no pending ACKs */
 713                 _debug("CLEAR ACKR TIMER");
 714                 del_timer_sync(&call->ackr_dfr_timo);
 715         }
 716
 717         /* send a special ACK if one is required */
 718         if (special_ACK) {
 719                 struct rxrpc_ackpacket ack;
 720                 struct kvec diov[2];
 721                 uint8_t acks[1] = { RXRPC_ACK_TYPE_ACK };
 722
 723                 /* fill out the appropriate form */
 724                 ack.bufferSpace = htons(RXRPC_CALL_ACK_WINDOW_SIZE);
 725                 ack.maxSkew     = htons(min(call->ackr_high_seq - seq,
 726                                             65535U));
 727                 ack.firstPacket = htonl(call->ackr_win_bot);
 728                 ack.previousPacket = call->ackr_prev_seq;
 729                 ack.serial      = hdr->serial;
 730                 ack.reason      = special_ACK;
 731                 ack.nAcks       = 0;
 732
 733                 _proto("Rx Sending s-ACK"
 734                        " { m=%hu f=#%u p=#%u s=%%%u r=%s n=%u }",
 735                        ntohs(ack.maxSkew),
 736                        ntohl(ack.firstPacket),
 737                        ntohl(ack.previousPacket),
 738                        ntohl(ack.serial),
 739                        rxrpc_acks[ack.reason],
 740                        ack.nAcks);
 741
 742                 diov[0].iov_len  = sizeof(struct rxrpc_ackpacket);
 743                 diov[0].iov_base = &ack;
 744                 diov[1].iov_len  = sizeof(acks);
 745                 diov[1].iov_base = acks;
 746
 747                 /* build and send the message */
 748                 err = rxrpc_conn_newmsg(call->conn,call, RXRPC_PACKET_TYPE_ACK,
 749                                         hdr->seq ? 2 : 1, diov,
 750                                         GFP_KERNEL,
 751                                         &msg);
 752                 if (err < 0) {
 753                         ret = err;
 754                         goto out;
 755                 }
 756
 757                 msg->seq = seq;
 758                 msg->hdr.seq = htonl(seq);
 759                 msg->hdr.flags |= RXRPC_SLOW_START_OK;
 760
 761                 err = rxrpc_conn_sendmsg(call->conn, msg);
 762                 rxrpc_put_message(msg);
 763                 if (err < 0) {
 764                         ret = err;
 765                         goto out;
 766                 }
 767                 call->pkt_snd_count++;
 768         }
 769
 770  out:
 771         if (hdr->seq)
 772                 call->ackr_prev_seq = hdr->seq;
 773
 774         _leave(" = %d", ret);
 775         return ret;
 776 } /* end rxrpc_call_generate_ACK() */
 777
 778 /*****************************************************************************/
 779 /*
 780  * handle work to be done on a call
 781  * - includes packet reception and timeout processing
 782  */
 783 void rxrpc_call_do_stuff(struct rxrpc_call *call)
 784 {
 785         _enter("%p{flags=%lx}", call, call->flags);
 786
 787         /* handle packet reception */
 788         if (call->flags & RXRPC_CALL_RCV_PKT) {
 789                 _debug("- receive packet");
 790                 call->flags &= ~RXRPC_CALL_RCV_PKT;
 791                 rxrpc_call_receive_packet(call);
 792         }
 793
 794         /* handle overdue ACKs */
 795         if (call->flags & RXRPC_CALL_ACKS_TIMO) {
 796                 _debug("- overdue ACK timeout");
 797                 call->flags &= ~RXRPC_CALL_ACKS_TIMO;
 798                 rxrpc_call_resend(call, call->snd_seq_count);
 799         }
 800
 801         /* handle lack of reception */
 802         if (call->flags & RXRPC_CALL_RCV_TIMO) {
 803                 _debug("- reception timeout");
 804                 call->flags &= ~RXRPC_CALL_RCV_TIMO;
 805                 rxrpc_call_abort(call, -EIO);
 806         }
 807
 808         /* handle deferred ACKs */
 809         if (call->flags & RXRPC_CALL_ACKR_TIMO ||
 810             (call->ackr.nAcks > 0 && call->ackr.reason == RXRPC_ACK_REQUESTED)
 811             ) {
 812                 _debug("- deferred ACK timeout: cj=%05lu r=%s n=%u",
 813                        jiffies - call->cjif,
 814                        rxrpc_acks[call->ackr.reason],
 815                        call->ackr.nAcks);
 816
 817                 call->flags &= ~RXRPC_CALL_ACKR_TIMO;
 818
 819                 if (call->ackr.nAcks > 0 &&
 820                     call->app_call_state != RXRPC_CSTATE_ERROR) {
 821                         /* generate ACK */
 822                         __rxrpc_call_gen_normal_ACK(call, call->ackr_dfr_seq);
 823                         call->ackr_dfr_seq = 0;
 824                 }
 825         }
 826
 827         _leave("");
 828
 829 } /* end rxrpc_call_do_stuff() */
 830
 831 /*****************************************************************************/
 832 /*
 833  * send an abort message at call or connection level
 834  * - must be called with call->lock held
 835  * - the supplied error code is sent as the packet data
 836  */
 837 static int __rxrpc_call_abort(struct rxrpc_call *call, int errno)
 838 {
 839         struct rxrpc_connection *conn = call->conn;
 840         struct rxrpc_message *msg;
 841         struct kvec diov[1];
 842         int ret;
 843         __be32 _error;
 844
 845         _enter("%p{%08x},%p{%d},%d",
 846                conn, ntohl(conn->conn_id), call, ntohl(call->call_id), errno);
 847
 848         /* if this call is already aborted, then just wake up any waiters */
 849         if (call->app_call_state == RXRPC_CSTATE_ERROR) {
 850                 spin_unlock(&call->lock);
 851                 call->app_error_func(call);
 852                 _leave(" = 0");
 853                 return 0;
 854         }
 855
 856         rxrpc_get_call(call);
 857
 858         /* change the state _with_ the lock still held */
 859         call->app_call_state    = RXRPC_CSTATE_ERROR;
 860         call->app_err_state     = RXRPC_ESTATE_LOCAL_ABORT;
 861         call->app_errno         = errno;
 862         call->app_mark          = RXRPC_APP_MARK_EOF;
 863         call->app_read_buf      = NULL;
 864         call->app_async_read    = 0;
 865
 866         _state(call);
 867
 868         /* ask the app to translate the error code */
 869         call->app_aemap_func(call);
 870
 871         spin_unlock(&call->lock);
 872
 873         /* flush any outstanding ACKs */
 874         del_timer_sync(&call->acks_timeout);
 875         del_timer_sync(&call->rcv_timeout);
 876         del_timer_sync(&call->ackr_dfr_timo);
 877
 878         if (rxrpc_call_is_ack_pending(call))
 879                 __rxrpc_call_gen_normal_ACK(call, 0);
 880
 881         /* send the abort packet only if we actually traded some other
 882          * packets */
 883         ret = 0;
 884         if (call->pkt_snd_count || call->pkt_rcv_count) {
 885                 /* actually send the abort */
 886                 _proto("Rx Sending Call ABORT { data=%d }",
 887                        call->app_abort_code);
 888
 889                 _error = htonl(call->app_abort_code);
 890
 891                 diov[0].iov_len  = sizeof(_error);
 892                 diov[0].iov_base = &_error;
 893
 894                 ret = rxrpc_conn_newmsg(conn, call, RXRPC_PACKET_TYPE_ABORT,
 895                                         1, diov, GFP_KERNEL, &msg);
 896                 if (ret == 0) {
 897                         ret = rxrpc_conn_sendmsg(conn, msg);
 898                         rxrpc_put_message(msg);
 899                 }
 900         }
 901
 902         /* tell the app layer to let go */
 903         call->app_error_func(call);
 904
 905         rxrpc_put_call(call);
 906
 907         _leave(" = %d", ret);
 908         return ret;
 909 } /* end __rxrpc_call_abort() */
 910
 911 /*****************************************************************************/
 912 /*
 913  * send an abort message at call or connection level
 914  * - the supplied error code is sent as the packet data
 915  */
 916 int rxrpc_call_abort(struct rxrpc_call *call, int error)
 917 {
 918         spin_lock(&call->lock);
 919
 920         return __rxrpc_call_abort(call, error);
 921
 922 } /* end rxrpc_call_abort() */
 923
 924 /*****************************************************************************/
 925 /*
 926  * process packets waiting for this call
 927  */
 928 static void rxrpc_call_receive_packet(struct rxrpc_call *call)
 929 {
 930         struct rxrpc_message *msg;
 931         struct list_head *_p;
 932
 933         _enter("%p", call);
 934
 935         rxrpc_get_call(call); /* must not go away too soon if aborted by
 936                                * app-layer */
 937
 938         while (!list_empty(&call->rcv_receiveq)) {
 939                 /* try to get next packet */
 940                 _p = NULL;
 941                 spin_lock(&call->lock);
 942                 if (!list_empty(&call->rcv_receiveq)) {
 943                         _p = call->rcv_receiveq.next;
 944                         list_del_init(_p);
 945                 }
 946                 spin_unlock(&call->lock);
 947
 948                 if (!_p)
 949                         break;
 950
 951                 msg = list_entry(_p, struct rxrpc_message, link);
 952
 953                 _proto("Rx %05lu Received %s packet (%%%u,#%u,%c%c%c%c%c)",
 954                        jiffies - call->cjif,
 955                        rxrpc_pkts[msg->hdr.type],
 956                        ntohl(msg->hdr.serial),
 957                        msg->seq,
 958                        msg->hdr.flags & RXRPC_JUMBO_PACKET      ? 'j' : '-',
 959                        msg->hdr.flags & RXRPC_MORE_PACKETS      ? 'm' : '-',
 960                        msg->hdr.flags & RXRPC_LAST_PACKET       ? 'l' : '-',
 961                        msg->hdr.flags & RXRPC_REQUEST_ACK       ? 'r' : '-',
 962                        msg->hdr.flags & RXRPC_CLIENT_INITIATED  ? 'C' : 'S'
 963                        );
 964
 965                 switch (msg->hdr.type) {
 966                         /* deal with data packets */
 967                 case RXRPC_PACKET_TYPE_DATA:
 968                         /* ACK the packet if necessary */
 969                         switch (rxrpc_call_generate_ACK(call, &msg->hdr,
 970                                                         NULL)) {
 971                         case 0: /* useful packet */
 972                                 rxrpc_call_receive_data_packet(call, msg);
 973                                 break;
 974                         case 1: /* duplicate or out-of-window packet */
 975                                 break;
 976                         default:
 977                                 rxrpc_put_message(msg);
 978                                 goto out;
 979                         }
 980                         break;
 981
 982                         /* deal with ACK packets */
 983                 case RXRPC_PACKET_TYPE_ACK:
 984                         rxrpc_call_receive_ack_packet(call, msg);
 985                         break;
 986
 987                         /* deal with abort packets */
 988                 case RXRPC_PACKET_TYPE_ABORT: {
 989                         __be32 _dbuf, *dp;
 990
 991                         dp = skb_header_pointer(msg->pkt, msg->offset,
 992                                                 sizeof(_dbuf), &_dbuf);
 993                         if (dp == NULL)
 994                                 printk("Rx Received short ABORT packet\n");
 995
 996                         _proto("Rx Received Call ABORT { data=%d }",
 997                                (dp ? ntohl(*dp) : 0));
 998
 999                         spin_lock(&call->lock);
1000                         call->app_call_state    = RXRPC_CSTATE_ERROR;
1001                         call->app_err_state     = RXRPC_ESTATE_PEER_ABORT;
1002                         call->app_abort_code    = (dp ? ntohl(*dp) : 0);
1003                         call->app_errno         = -ECONNABORTED;
1004                         call->app_mark          = RXRPC_APP_MARK_EOF;
1005                         call->app_read_buf      = NULL;
1006                         call->app_async_read    = 0;
1007
1008                         /* ask the app to translate the error code */
1009                         call->app_aemap_func(call);
1010                         _state(call);
1011                         spin_unlock(&call->lock);
1012                         call->app_error_func(call);
1013                         break;
1014                 }
1015                 default:
1016                         /* deal with other packet types */
1017                         _proto("Rx Unsupported packet type %u (#%u)",
1018                                msg->hdr.type, msg->seq);
1019                         break;
1020                 }
1021
1022                 rxrpc_put_message(msg);
1023         }
1024
1025  out:
1026         rxrpc_put_call(call);
1027         _leave("");
1028 } /* end rxrpc_call_receive_packet() */
1029
1030 /*****************************************************************************/
1031 /*
1032  * process next data packet
1033  * - as the next data packet arrives:
1034  *   - it is queued on app_readyq _if_ it is the next one expected
1035  *     (app_ready_seq+1)
1036  *   - it is queued on app_unreadyq _if_ it is not the next one expected
1037  *   - if a packet placed on app_readyq completely fills a hole leading up to
1038  *     the first packet on app_unreadyq, then packets now in sequence are
1039  *     tranferred to app_readyq
1040  * - the application layer can only see packets on app_readyq
1041  *   (app_ready_qty bytes)
1042  * - the application layer is prodded every time a new packet arrives
1043  */
1044 static void rxrpc_call_receive_data_packet(struct rxrpc_call *call,
1045                                            struct rxrpc_message *msg)
1046 {
1047         const struct rxrpc_operation *optbl, *op;
1048         struct rxrpc_message *pmsg;
1049         struct list_head *_p;
1050         int ret, lo, hi, rmtimo;
1051         __be32 opid;
1052
1053         _enter("%p{%u},%p{%u}", call, ntohl(call->call_id), msg, msg->seq);
1054
1055         rxrpc_get_message(msg);
1056
1057         /* add to the unready queue if we'd have to create a hole in the ready
1058          * queue otherwise */
1059         if (msg->seq != call->app_ready_seq + 1) {
1060                 _debug("Call add packet %d to unreadyq", msg->seq);
1061
1062                 /* insert in seq order */
1063                 list_for_each(_p, &call->app_unreadyq) {
1064                         pmsg = list_entry(_p, struct rxrpc_message, link);
1065                         if (pmsg->seq > msg->seq)
1066                                 break;
1067                 }
1068
1069                 list_add_tail(&msg->link, _p);
1070
1071                 _leave(" [unreadyq]");
1072                 return;
1073         }
1074
1075         /* next in sequence - simply append into the call's ready queue */
1076         _debug("Call add packet %d to readyq (+%Zd => %Zd bytes)",
1077                msg->seq, msg->dsize, call->app_ready_qty);
1078
1079         spin_lock(&call->lock);
1080         call->app_ready_seq = msg->seq;
1081         call->app_ready_qty += msg->dsize;
1082         list_add_tail(&msg->link, &call->app_readyq);
1083
1084         /* move unready packets to the readyq if we got rid of a hole */
1085         while (!list_empty(&call->app_unreadyq)) {
1086                 pmsg = list_entry(call->app_unreadyq.next,
1087                                   struct rxrpc_message, link);
1088
1089                 if (pmsg->seq != call->app_ready_seq + 1)
1090                         break;
1091
1092                 /* next in sequence - just move list-to-list */
1093                 _debug("Call transfer packet %d to readyq (+%Zd => %Zd bytes)",
1094                        pmsg->seq, pmsg->dsize, call->app_ready_qty);
1095
1096                 call->app_ready_seq = pmsg->seq;
1097                 call->app_ready_qty += pmsg->dsize;
1098                 list_del_init(&pmsg->link);
1099                 list_add_tail(&pmsg->link, &call->app_readyq);
1100         }
1101
1102         /* see if we've got the last packet yet */
1103         if (!list_empty(&call->app_readyq)) {
1104                 pmsg = list_entry(call->app_readyq.prev,
1105                                   struct rxrpc_message, link);
1106                 if (pmsg->hdr.flags & RXRPC_LAST_PACKET) {
1107                         call->app_last_rcv = 1;
1108                         _debug("Last packet on readyq");
1109                 }
1110         }
1111
1112         switch (call->app_call_state) {
1113                 /* do nothing if call already aborted */
1114         case RXRPC_CSTATE_ERROR:
1115                 spin_unlock(&call->lock);
1116                 _leave(" [error]");
1117                 return;
1118
1119                 /* extract the operation ID from an incoming call if that's not
1120                  * yet been done */
1121         case RXRPC_CSTATE_SRVR_RCV_OPID:
1122                 spin_unlock(&call->lock);
1123
1124                 /* handle as yet insufficient data for the operation ID */
1125                 if (call->app_ready_qty < 4) {
1126                         if (call->app_last_rcv)
1127                                 /* trouble - last packet seen */
1128                                 rxrpc_call_abort(call, -EINVAL);
1129
1130                         _leave("");
1131                         return;
1132                 }
1133
1134                 /* pull the operation ID out of the buffer */
1135                 ret = rxrpc_call_read_data(call, &opid, sizeof(opid), 0);
1136                 if (ret < 0) {
1137                         printk("Unexpected error from read-data: %d\n", ret);
1138                         if (call->app_call_state != RXRPC_CSTATE_ERROR)
1139                                 rxrpc_call_abort(call, ret);
1140                         _leave("");
1141                         return;
1142                 }
1143                 call->app_opcode = ntohl(opid);
1144
1145                 /* locate the operation in the available ops table */
1146                 optbl = call->conn->service->ops_begin;
1147                 lo = 0;
1148                 hi = call->conn->service->ops_end - optbl;
1149
1150                 while (lo < hi) {
1151                         int mid = (hi + lo) / 2;
1152                         op = &optbl[mid];
1153                         if (call->app_opcode == op->id)
1154                                 goto found_op;
1155                         if (call->app_opcode > op->id)
1156                                 lo = mid + 1;
1157                         else
1158                                 hi = mid;
1159                 }
1160
1161                 /* search failed */
1162                 kproto("Rx Client requested operation %d from %s service",
1163                        call->app_opcode, call->conn->service->name);
1164                 rxrpc_call_abort(call, -EINVAL);
1165                 _leave(" [inval]");
1166                 return;
1167
1168         found_op:
1169                 _proto("Rx Client requested operation %s from %s service",
1170                        op->name, call->conn->service->name);
1171
1172                 /* we're now waiting for the argument block (unless the call
1173                  * was aborted) */
1174                 spin_lock(&call->lock);
1175                 if (call->app_call_state == RXRPC_CSTATE_SRVR_RCV_OPID ||
1176                     call->app_call_state == RXRPC_CSTATE_SRVR_SND_REPLY) {
1177                         if (!call->app_last_rcv)
1178                                 call->app_call_state =
1179                                         RXRPC_CSTATE_SRVR_RCV_ARGS;
1180                         else if (call->app_ready_qty > 0)
1181                                 call->app_call_state =
1182                                         RXRPC_CSTATE_SRVR_GOT_ARGS;
1183                         else
1184                                 call->app_call_state =
1185                                         RXRPC_CSTATE_SRVR_SND_REPLY;
1186                         call->app_mark = op->asize;
1187                         call->app_user = op->user;
1188                 }
1189                 spin_unlock(&call->lock);
1190
1191                 _state(call);
1192                 break;
1193
1194         case RXRPC_CSTATE_SRVR_RCV_ARGS:
1195                 /* change state if just received last packet of arg block */
1196                 if (call->app_last_rcv)
1197                         call->app_call_state = RXRPC_CSTATE_SRVR_GOT_ARGS;
1198                 spin_unlock(&call->lock);
1199
1200                 _state(call);
1201                 break;
1202
1203         case RXRPC_CSTATE_CLNT_RCV_REPLY:
1204                 /* change state if just received last packet of reply block */
1205                 rmtimo = 0;
1206                 if (call->app_last_rcv) {
1207                         call->app_call_state = RXRPC_CSTATE_CLNT_GOT_REPLY;
1208                         rmtimo = 1;
1209                 }
1210                 spin_unlock(&call->lock);
1211
1212                 if (rmtimo) {
1213                         del_timer_sync(&call->acks_timeout);
1214                         del_timer_sync(&call->rcv_timeout);
1215                         del_timer_sync(&call->ackr_dfr_timo);
1216                 }
1217
1218                 _state(call);
1219                 break;
1220
1221         default:
1222                 /* deal with data reception in an unexpected state */
1223                 printk("Unexpected state [[[ %u ]]]\n", call->app_call_state);
1224                 __rxrpc_call_abort(call, -EBADMSG);
1225                 _leave("");
1226                 return;
1227         }
1228
1229         if (call->app_call_state == RXRPC_CSTATE_CLNT_RCV_REPLY &&
1230             call->app_last_rcv)
1231                 BUG();
1232
1233         /* otherwise just invoke the data function whenever we can satisfy its desire for more
1234          * data
1235          */
1236         _proto("Rx Received Op Data: st=%u qty=%Zu mk=%Zu%s",
1237                call->app_call_state, call->app_ready_qty, call->app_mark,
1238                call->app_last_rcv ? " last-rcvd" : "");
1239
1240         spin_lock(&call->lock);
1241
1242         ret = __rxrpc_call_read_data(call);
1243         switch (ret) {
1244         case 0:
1245                 spin_unlock(&call->lock);
1246                 call->app_attn_func(call);
1247                 break;
1248         case -EAGAIN:
1249                 spin_unlock(&call->lock);
1250                 break;
1251         case -ECONNABORTED:
1252                 spin_unlock(&call->lock);
1253                 break;
1254         default:
1255                 __rxrpc_call_abort(call, ret);
1256                 break;
1257         }
1258
1259         _state(call);
1260
1261         _leave("");
1262
1263 } /* end rxrpc_call_receive_data_packet() */
1264
1265 /*****************************************************************************/
1266 /*
1267  * received an ACK packet
1268  */
1269 static void rxrpc_call_receive_ack_packet(struct rxrpc_call *call,
1270                                           struct rxrpc_message *msg)
1271 {
1272         struct rxrpc_ackpacket _ack, *ap;
1273         rxrpc_serial_net_t serial;
1274         rxrpc_seq_t seq;
1275         int ret;
1276
1277         _enter("%p{%u},%p{%u}", call, ntohl(call->call_id), msg, msg->seq);
1278
1279         /* extract the basic ACK record */
1280         ap = skb_header_pointer(msg->pkt, msg->offset, sizeof(_ack), &_ack);
1281         if (ap == NULL) {
1282                 printk("Rx Received short ACK packet\n");
1283                 return;
1284         }
1285         msg->offset += sizeof(_ack);
1286
1287         serial = ap->serial;
1288         seq = ntohl(ap->firstPacket);
1289
1290         _proto("Rx Received ACK %%%d { b=%hu m=%hu f=%u p=%u s=%u r=%s n=%u }",
1291                ntohl(msg->hdr.serial),
1292                ntohs(ap->bufferSpace),
1293                ntohs(ap->maxSkew),
1294                seq,
1295                ntohl(ap->previousPacket),
1296                ntohl(serial),
1297                rxrpc_acks[ap->reason],
1298                call->ackr.nAcks
1299                );
1300
1301         /* check the other side isn't ACK'ing a sequence number I haven't sent
1302          * yet */
1303         if (ap->nAcks > 0 &&
1304             (seq > call->snd_seq_count ||
1305              seq + ap->nAcks - 1 > call->snd_seq_count)) {
1306                 printk("Received ACK (#%u-#%u) for unsent packet\n",
1307                        seq, seq + ap->nAcks - 1);
1308                 rxrpc_call_abort(call, -EINVAL);
1309                 _leave("");
1310                 return;
1311         }
1312
1313         /* deal with RTT calculation */
1314         if (serial) {
1315                 struct rxrpc_message *rttmsg;
1316
1317                 /* find the prompting packet */
1318                 spin_lock(&call->lock);
1319                 if (call->snd_ping && call->snd_ping->hdr.serial == serial) {
1320                         /* it was a ping packet */
1321                         rttmsg = call->snd_ping;
1322                         call->snd_ping = NULL;
1323                         spin_unlock(&call->lock);
1324
1325                         if (rttmsg) {
1326                                 rttmsg->rttdone = 1;
1327                                 rxrpc_peer_calculate_rtt(call->conn->peer,
1328                                                          rttmsg, msg);
1329                                 rxrpc_put_message(rttmsg);
1330                         }
1331                 }
1332                 else {
1333                         struct list_head *_p;
1334
1335                         /* it ought to be a data packet - look in the pending
1336                          * ACK list */
1337                         list_for_each(_p, &call->acks_pendq) {
1338                                 rttmsg = list_entry(_p, struct rxrpc_message,
1339                                                     link);
1340                                 if (rttmsg->hdr.serial == serial) {
1341                                         if (rttmsg->rttdone)
1342                                                 /* never do RTT twice without
1343                                                  * resending */
1344                                                 break;
1345
1346                                         rttmsg->rttdone = 1;
1347                                         rxrpc_peer_calculate_rtt(
1348                                                 call->conn->peer, rttmsg, msg);
1349                                         break;
1350                                 }
1351                         }
1352                         spin_unlock(&call->lock);
1353                 }
1354         }
1355
1356         switch (ap->reason) {
1357                 /* deal with negative/positive acknowledgement of data
1358                  * packets */
1359         case RXRPC_ACK_REQUESTED:
1360         case RXRPC_ACK_DELAY:
1361         case RXRPC_ACK_IDLE:
1362                 rxrpc_call_definitively_ACK(call, seq - 1);
1363
1364         case RXRPC_ACK_DUPLICATE:
1365         case RXRPC_ACK_OUT_OF_SEQUENCE:
1366         case RXRPC_ACK_EXCEEDS_WINDOW:
1367                 call->snd_resend_cnt = 0;
1368                 ret = rxrpc_call_record_ACK(call, msg, seq, ap->nAcks);
1369                 if (ret < 0)
1370                         rxrpc_call_abort(call, ret);
1371                 break;
1372
1373                 /* respond to ping packets immediately */
1374         case RXRPC_ACK_PING:
1375                 rxrpc_call_generate_ACK(call, &msg->hdr, ap);
1376                 break;
1377
1378                 /* only record RTT on ping response packets */
1379         case RXRPC_ACK_PING_RESPONSE:
1380                 if (call->snd_ping) {
1381                         struct rxrpc_message *rttmsg;
1382
1383                         /* only do RTT stuff if the response matches the
1384                          * retained ping */
1385                         rttmsg = NULL;
1386                         spin_lock(&call->lock);
1387                         if (call->snd_ping &&
1388                             call->snd_ping->hdr.serial == ap->serial) {
1389                                 rttmsg = call->snd_ping;
1390                                 call->snd_ping = NULL;
1391                         }
1392                         spin_unlock(&call->lock);
1393
1394                         if (rttmsg) {
1395                                 rttmsg->rttdone = 1;
1396                                 rxrpc_peer_calculate_rtt(call->conn->peer,
1397                                                          rttmsg, msg);
1398                                 rxrpc_put_message(rttmsg);
1399                         }
1400                 }
1401                 break;
1402
1403         default:
1404                 printk("Unsupported ACK reason %u\n", ap->reason);
1405                 break;
1406         }
1407
1408         _leave("");
1409 } /* end rxrpc_call_receive_ack_packet() */
1410
1411 /*****************************************************************************/
1412 /*
1413  * record definitive ACKs for all messages up to and including the one with the
1414  * 'highest' seq
1415  */
1416 static void rxrpc_call_definitively_ACK(struct rxrpc_call *call,
1417                                         rxrpc_seq_t highest)
1418 {
1419         struct rxrpc_message *msg;
1420         int now_complete;
1421
1422         _enter("%p{ads=%u},%u", call, call->acks_dftv_seq, highest);
1423
1424         while (call->acks_dftv_seq < highest) {
1425                 call->acks_dftv_seq++;
1426
1427                 _proto("Definitive ACK on packet #%u", call->acks_dftv_seq);
1428
1429                 /* discard those at front of queue until message with highest
1430                  * ACK is found */
1431                 spin_lock(&call->lock);
1432                 msg = NULL;
1433                 if (!list_empty(&call->acks_pendq)) {
1434                         msg = list_entry(call->acks_pendq.next,
1435                                          struct rxrpc_message, link);
1436                         list_del_init(&msg->link); /* dequeue */
1437                         if (msg->state == RXRPC_MSG_SENT)
1438                                 call->acks_pend_cnt--;
1439                 }
1440                 spin_unlock(&call->lock);
1441
1442                 /* insanity check */
1443                 if (!msg)
1444                         panic("%s(): acks_pendq unexpectedly empty\n",
1445                               __FUNCTION__);
1446
1447                 if (msg->seq != call->acks_dftv_seq)
1448                         panic("%s(): Packet #%u expected at front of acks_pendq"
1449                               " (#%u found)\n",
1450                               __FUNCTION__, call->acks_dftv_seq, msg->seq);
1451
1452                 /* discard the message */
1453                 msg->state = RXRPC_MSG_DONE;
1454                 rxrpc_put_message(msg);
1455         }
1456
1457         /* if all sent packets are definitively ACK'd then prod any sleepers just in case */
1458         now_complete = 0;
1459         spin_lock(&call->lock);
1460         if (call->acks_dftv_seq == call->snd_seq_count) {
1461                 if (call->app_call_state != RXRPC_CSTATE_COMPLETE) {
1462                         call->app_call_state = RXRPC_CSTATE_COMPLETE;
1463                         _state(call);
1464                         now_complete = 1;
1465                 }
1466         }
1467         spin_unlock(&call->lock);
1468
1469         if (now_complete) {
1470                 del_timer_sync(&call->acks_timeout);
1471                 del_timer_sync(&call->rcv_timeout);
1472                 del_timer_sync(&call->ackr_dfr_timo);
1473                 call->app_attn_func(call);
1474         }
1475
1476         _leave("");
1477 } /* end rxrpc_call_definitively_ACK() */
1478
1479 /*****************************************************************************/
1480 /*
1481  * record the specified amount of ACKs/NAKs
1482  */
1483 static int rxrpc_call_record_ACK(struct rxrpc_call *call,
1484                                  struct rxrpc_message *msg,
1485                                  rxrpc_seq_t seq,
1486                                  size_t count)
1487 {
1488         struct rxrpc_message *dmsg;
1489         struct list_head *_p;
1490         rxrpc_seq_t highest;
1491         unsigned ix;
1492         size_t chunk;
1493         char resend, now_complete;
1494         u8 acks[16];
1495
1496         _enter("%p{apc=%u ads=%u},%p,%u,%Zu",
1497                call, call->acks_pend_cnt, call->acks_dftv_seq,
1498                msg, seq, count);
1499
1500         /* handle re-ACK'ing of definitively ACK'd packets (may be out-of-order
1501          * ACKs) */
1502         if (seq <= call->acks_dftv_seq) {
1503                 unsigned delta = call->acks_dftv_seq - seq;
1504
1505                 if (count <= delta) {
1506                         _leave(" = 0 [all definitively ACK'd]");
1507                         return 0;
1508                 }
1509
1510                 seq += delta;
1511                 count -= delta;
1512                 msg->offset += delta;
1513         }
1514
1515         highest = seq + count - 1;
1516         resend = 0;
1517         while (count > 0) {
1518                 /* extract up to 16 ACK slots at a time */
1519                 chunk = min(count, sizeof(acks));
1520                 count -= chunk;
1521
1522                 memset(acks, 2, sizeof(acks));
1523
1524                 if (skb_copy_bits(msg->pkt, msg->offset, &acks, chunk) < 0) {
1525                         printk("Rx Received short ACK packet\n");
1526                         _leave(" = -EINVAL");
1527                         return -EINVAL;
1528                 }
1529                 msg->offset += chunk;
1530
1531                 /* check that the ACK set is valid */
1532                 for (ix = 0; ix < chunk; ix++) {
1533                         switch (acks[ix]) {
1534                         case RXRPC_ACK_TYPE_ACK:
1535                                 break;
1536                         case RXRPC_ACK_TYPE_NACK:
1537                                 resend = 1;
1538                                 break;
1539                         default:
1540                                 printk("Rx Received unsupported ACK state"
1541                                        " %u\n", acks[ix]);
1542                                 _leave(" = -EINVAL");
1543                                 return -EINVAL;
1544                         }
1545                 }
1546
1547                 _proto("Rx ACK of packets #%u-#%u "
1548                        "[%c%c%c%c%c%c%c%c%c%c%c%c%c%c%c%c] (pend=%u)",
1549                        seq, (unsigned) (seq + chunk - 1),
1550                        _acktype[acks[0x0]],
1551                        _acktype[acks[0x1]],
1552                        _acktype[acks[0x2]],
1553                        _acktype[acks[0x3]],
1554                        _acktype[acks[0x4]],
1555                        _acktype[acks[0x5]],
1556                        _acktype[acks[0x6]],
1557                        _acktype[acks[0x7]],
1558                        _acktype[acks[0x8]],
1559                        _acktype[acks[0x9]],
1560                        _acktype[acks[0xA]],
1561                        _acktype[acks[0xB]],
1562                        _acktype[acks[0xC]],
1563                        _acktype[acks[0xD]],
1564                        _acktype[acks[0xE]],
1565                        _acktype[acks[0xF]],
1566                        call->acks_pend_cnt
1567                        );
1568
1569                 /* mark the packets in the ACK queue as being provisionally
1570                  * ACK'd */
1571                 ix = 0;
1572                 spin_lock(&call->lock);
1573
1574                 /* find the first packet ACK'd/NAK'd here */
1575                 list_for_each(_p, &call->acks_pendq) {
1576                         dmsg = list_entry(_p, struct rxrpc_message, link);
1577                         if (dmsg->seq == seq)
1578                                 goto found_first;
1579                         _debug("- %u: skipping #%u", ix, dmsg->seq);
1580                 }
1581                 goto bad_queue;
1582
1583         found_first:
1584                 do {
1585                         _debug("- %u: processing #%u (%c) apc=%u",
1586                                ix, dmsg->seq, _acktype[acks[ix]],
1587                                call->acks_pend_cnt);
1588
1589                         if (acks[ix] == RXRPC_ACK_TYPE_ACK) {
1590                                 if (dmsg->state == RXRPC_MSG_SENT)
1591                                         call->acks_pend_cnt--;
1592                                 dmsg->state = RXRPC_MSG_ACKED;
1593                         }
1594                         else {
1595                                 if (dmsg->state == RXRPC_MSG_ACKED)
1596                                         call->acks_pend_cnt++;
1597                                 dmsg->state = RXRPC_MSG_SENT;
1598                         }
1599                         ix++;
1600                         seq++;
1601
1602                         _p = dmsg->link.next;
1603                         dmsg = list_entry(_p, struct rxrpc_message, link);
1604                 } while(ix < chunk &&
1605                         _p != &call->acks_pendq &&
1606                         dmsg->seq == seq);
1607
1608                 if (ix < chunk)
1609                         goto bad_queue;
1610
1611                 spin_unlock(&call->lock);
1612         }
1613
1614         if (resend)
1615                 rxrpc_call_resend(call, highest);
1616
1617         /* if all packets are provisionally ACK'd, then wake up anyone who's
1618          * waiting for that */
1619         now_complete = 0;
1620         spin_lock(&call->lock);
1621         if (call->acks_pend_cnt == 0) {
1622                 if (call->app_call_state == RXRPC_CSTATE_SRVR_RCV_FINAL_ACK) {
1623                         call->app_call_state = RXRPC_CSTATE_COMPLETE;
1624                         _state(call);
1625                 }
1626                 now_complete = 1;
1627         }
1628         spin_unlock(&call->lock);
1629
1630         if (now_complete) {
1631                 _debug("- wake up waiters");
1632                 del_timer_sync(&call->acks_timeout);
1633                 del_timer_sync(&call->rcv_timeout);
1634                 del_timer_sync(&call->ackr_dfr_timo);
1635                 call->app_attn_func(call);
1636         }
1637
1638         _leave(" = 0 (apc=%u)", call->acks_pend_cnt);
1639         return 0;
1640
1641  bad_queue:
1642         panic("%s(): acks_pendq in bad state (packet #%u absent)\n",
1643               __FUNCTION__, seq);
1644
1645 } /* end rxrpc_call_record_ACK() */
1646
1647 /*****************************************************************************/
1648 /*
1649  * transfer data from the ready packet queue to the asynchronous read buffer
1650  * - since this func is the only one going to look at packets queued on
1651  *   app_readyq, we don't need a lock to modify or access them, only to modify
1652  *   the queue pointers
1653  * - called with call->lock held
1654  * - the buffer must be in kernel space
1655  * - returns:
1656  *      0 if buffer filled
1657  *      -EAGAIN if buffer not filled and more data to come
1658  *      -EBADMSG if last packet received and insufficient data left
1659  *      -ECONNABORTED if the call has in an error state
1660  */
1661 static int __rxrpc_call_read_data(struct rxrpc_call *call)
1662 {
1663         struct rxrpc_message *msg;
1664         size_t qty;
1665         int ret;
1666
1667         _enter("%p{as=%d buf=%p qty=%Zu/%Zu}",
1668                call,
1669                call->app_async_read, call->app_read_buf,
1670                call->app_ready_qty, call->app_mark);
1671
1672         /* check the state */
1673         switch (call->app_call_state) {
1674         case RXRPC_CSTATE_SRVR_RCV_ARGS:
1675         case RXRPC_CSTATE_CLNT_RCV_REPLY:
1676                 if (call->app_last_rcv) {
1677                         printk("%s(%p,%p,%Zd):"
1678                                " Inconsistent call state (%s, last pkt)",
1679                                __FUNCTION__,
1680                                call, call->app_read_buf, call->app_mark,
1681                                rxrpc_call_states[call->app_call_state]);
1682                         BUG();
1683                 }
1684                 break;
1685
1686         case RXRPC_CSTATE_SRVR_RCV_OPID:
1687         case RXRPC_CSTATE_SRVR_GOT_ARGS:
1688         case RXRPC_CSTATE_CLNT_GOT_REPLY:
1689                 break;
1690
1691         case RXRPC_CSTATE_SRVR_SND_REPLY:
1692                 if (!call->app_last_rcv) {
1693                         printk("%s(%p,%p,%Zd):"
1694                                " Inconsistent call state (%s, not last pkt)",
1695                                __FUNCTION__,
1696                                call, call->app_read_buf, call->app_mark,
1697                                rxrpc_call_states[call->app_call_state]);
1698                         BUG();
1699                 }
1700                 _debug("Trying to read data from call in SND_REPLY state");
1701                 break;
1702
1703         case RXRPC_CSTATE_ERROR:
1704                 _leave(" = -ECONNABORTED");
1705                 return -ECONNABORTED;
1706
1707         default:
1708                 printk("reading in unexpected state [[[ %u ]]]\n",
1709                        call->app_call_state);
1710                 BUG();
1711         }
1712
1713         /* handle the case of not having an async buffer */
1714         if (!call->app_async_read) {
1715                 if (call->app_mark == RXRPC_APP_MARK_EOF) {
1716                         ret = call->app_last_rcv ? 0 : -EAGAIN;
1717                 }
1718                 else {
1719                         if (call->app_mark >= call->app_ready_qty) {
1720                                 call->app_mark = RXRPC_APP_MARK_EOF;
1721                                 ret = 0;
1722                         }
1723                         else {
1724                                 ret = call->app_last_rcv ? -EBADMSG : -EAGAIN;
1725                         }
1726                 }
1727
1728                 _leave(" = %d [no buf]", ret);
1729                 return 0;
1730         }
1731
1732         while (!list_empty(&call->app_readyq) && call->app_mark > 0) {
1733                 msg = list_entry(call->app_readyq.next,
1734                                  struct rxrpc_message, link);
1735
1736                 /* drag as much data as we need out of this packet */
1737                 qty = min(call->app_mark, msg->dsize);
1738
1739                 _debug("reading %Zu from skb=%p off=%lu",
1740                        qty, msg->pkt, msg->offset);
1741
1742                 if (call->app_read_buf)
1743                         if (skb_copy_bits(msg->pkt, msg->offset,
1744                                           call->app_read_buf, qty) < 0)
1745                                 panic("%s: Failed to copy data from packet:"
1746                                       " (%p,%p,%Zd)",
1747                                       __FUNCTION__,
1748                                       call, call->app_read_buf, qty);
1749
1750                 /* if that packet is now empty, discard it */
1751                 call->app_ready_qty -= qty;
1752                 msg->dsize -= qty;
1753
1754                 if (msg->dsize == 0) {
1755                         list_del_init(&msg->link);
1756                         rxrpc_put_message(msg);
1757                 }
1758                 else {
1759                         msg->offset += qty;
1760                 }
1761
1762                 call->app_mark -= qty;
1763                 if (call->app_read_buf)
1764                         call->app_read_buf += qty;
1765         }
1766
1767         if (call->app_mark == 0) {
1768                 call->app_async_read = 0;
1769                 call->app_mark = RXRPC_APP_MARK_EOF;
1770                 call->app_read_buf = NULL;
1771
1772                 /* adjust the state if used up all packets */
1773                 if (list_empty(&call->app_readyq) && call->app_last_rcv) {
1774                         switch (call->app_call_state) {
1775                         case RXRPC_CSTATE_SRVR_RCV_OPID:
1776                                 call->app_call_state = RXRPC_CSTATE_SRVR_SND_REPLY;
1777                                 call->app_mark = RXRPC_APP_MARK_EOF;
1778                                 _state(call);
1779                                 del_timer_sync(&call->rcv_timeout);
1780                                 break;
1781                         case RXRPC_CSTATE_SRVR_GOT_ARGS:
1782                                 call->app_call_state = RXRPC_CSTATE_SRVR_SND_REPLY;
1783                                 _state(call);
1784                                 del_timer_sync(&call->rcv_timeout);
1785                                 break;
1786                         default:
1787                                 call->app_call_state = RXRPC_CSTATE_COMPLETE;
1788                                 _state(call);
1789                                 del_timer_sync(&call->acks_timeout);
1790                                 del_timer_sync(&call->ackr_dfr_timo);
1791                                 del_timer_sync(&call->rcv_timeout);
1792                                 break;
1793                         }
1794                 }
1795
1796                 _leave(" = 0");
1797                 return 0;
1798         }
1799
1800         if (call->app_last_rcv) {
1801                 _debug("Insufficient data (%Zu/%Zu)",
1802                        call->app_ready_qty, call->app_mark);
1803                 call->app_async_read = 0;
1804                 call->app_mark = RXRPC_APP_MARK_EOF;
1805                 call->app_read_buf = NULL;
1806
1807                 _leave(" = -EBADMSG");
1808                 return -EBADMSG;
1809         }
1810
1811         _leave(" = -EAGAIN");
1812         return -EAGAIN;
1813 } /* end __rxrpc_call_read_data() */
1814
1815 /*****************************************************************************/
1816 /*
1817  * attempt to read the specified amount of data from the call's ready queue
1818  * into the buffer provided
1819  * - since this func is the only one going to look at packets queued on
1820  *   app_readyq, we don't need a lock to modify or access them, only to modify
1821  *   the queue pointers
1822  * - if the buffer pointer is NULL, then data is merely drained, not copied
1823  * - if flags&RXRPC_CALL_READ_BLOCK, then the function will wait until there is
1824  *   enough data or an error will be generated
1825  *   - note that the caller must have added the calling task to the call's wait
1826  *     queue beforehand
1827  * - if flags&RXRPC_CALL_READ_ALL, then an error will be generated if this
1828  *   function doesn't read all available data
1829  */
1830 int rxrpc_call_read_data(struct rxrpc_call *call,
1831                          void *buffer, size_t size, int flags)
1832 {
1833         int ret;
1834
1835         _enter("%p{arq=%Zu},%p,%Zd,%x",
1836                call, call->app_ready_qty, buffer, size, flags);
1837
1838         spin_lock(&call->lock);
1839
1840         if (unlikely(!!call->app_read_buf)) {
1841                 spin_unlock(&call->lock);
1842                 _leave(" = -EBUSY");
1843                 return -EBUSY;
1844         }
1845
1846         call->app_mark = size;
1847         call->app_read_buf = buffer;
1848         call->app_async_read = 1;
1849         call->app_read_count++;
1850
1851         /* read as much data as possible */
1852         ret = __rxrpc_call_read_data(call);
1853         switch (ret) {
1854         case 0:
1855                 if (flags & RXRPC_CALL_READ_ALL &&
1856                     (!call->app_last_rcv || call->app_ready_qty > 0)) {
1857                         _leave(" = -EBADMSG");
1858                         __rxrpc_call_abort(call, -EBADMSG);
1859                         return -EBADMSG;
1860                 }
1861
1862                 spin_unlock(&call->lock);
1863                 call->app_attn_func(call);
1864                 _leave(" = 0");
1865                 return ret;
1866
1867         case -ECONNABORTED:
1868                 spin_unlock(&call->lock);
1869                 _leave(" = %d [aborted]", ret);
1870                 return ret;
1871
1872         default:
1873                 __rxrpc_call_abort(call, ret);
1874                 _leave(" = %d", ret);
1875                 return ret;
1876
1877         case -EAGAIN:
1878                 spin_unlock(&call->lock);
1879
1880                 if (!(flags & RXRPC_CALL_READ_BLOCK)) {
1881                         _leave(" = -EAGAIN");
1882                         return -EAGAIN;
1883                 }
1884
1885                 /* wait for the data to arrive */
1886                 _debug("blocking for data arrival");
1887
1888                 for (;;) {
1889                         set_current_state(TASK_INTERRUPTIBLE);
1890                         if (!call->app_async_read || signal_pending(current))
1891                                 break;
1892                         schedule();
1893                 }
1894                 set_current_state(TASK_RUNNING);
1895
1896                 if (signal_pending(current)) {
1897                         _leave(" = -EINTR");
1898                         return -EINTR;
1899                 }
1900
1901                 if (call->app_call_state == RXRPC_CSTATE_ERROR) {
1902                         _leave(" = -ECONNABORTED");
1903                         return -ECONNABORTED;
1904                 }
1905
1906                 _leave(" = 0");
1907                 return 0;
1908         }
1909
1910 } /* end rxrpc_call_read_data() */
1911
1912 /*****************************************************************************/
1913 /*
1914  * write data to a call
1915  * - the data may not be sent immediately if it doesn't fill a buffer
1916  * - if we can't queue all the data for buffering now, siov[] will have been
1917  *   adjusted to take account of what has been sent
1918  */
1919 int rxrpc_call_write_data(struct rxrpc_call *call,
1920                           size_t sioc,
1921                           struct kvec *siov,
1922                           u8 rxhdr_flags,
1923                           int alloc_flags,
1924                           int dup_data,
1925                           size_t *size_sent)
1926 {
1927         struct rxrpc_message *msg;
1928         struct kvec *sptr;
1929         size_t space, size, chunk, tmp;
1930         char *buf;
1931         int ret;
1932
1933         _enter("%p,%Zu,%p,%02x,%x,%d,%p",
1934                call, sioc, siov, rxhdr_flags, alloc_flags, dup_data,
1935                size_sent);
1936
1937         *size_sent = 0;
1938         size = 0;
1939         ret = -EINVAL;
1940
1941         /* can't send more if we've sent last packet from this end */
1942         switch (call->app_call_state) {
1943         case RXRPC_CSTATE_SRVR_SND_REPLY:
1944         case RXRPC_CSTATE_CLNT_SND_ARGS:
1945                 break;
1946         case RXRPC_CSTATE_ERROR:
1947                 ret = call->app_errno;
1948         default:
1949                 goto out;
1950         }
1951
1952         /* calculate how much data we've been given */
1953         sptr = siov;
1954         for (; sioc > 0; sptr++, sioc--) {
1955                 if (!sptr->iov_len)
1956                         continue;
1957
1958                 if (!sptr->iov_base)
1959                         goto out;
1960
1961                 size += sptr->iov_len;
1962         }
1963
1964         _debug("- size=%Zu mtu=%Zu", size, call->conn->mtu_size);
1965
1966         do {
1967                 /* make sure there's a message under construction */
1968                 if (!call->snd_nextmsg) {
1969                         /* no - allocate a message with no data yet attached */
1970                         ret = rxrpc_conn_newmsg(call->conn, call,
1971                                                 RXRPC_PACKET_TYPE_DATA,
1972                                                 0, NULL, alloc_flags,
1973                                                 &call->snd_nextmsg);
1974                         if (ret < 0)
1975                                 goto out;
1976                         _debug("- allocated new message [ds=%Zu]",
1977                                call->snd_nextmsg->dsize);
1978                 }
1979
1980                 msg = call->snd_nextmsg;
1981                 msg->hdr.flags |= rxhdr_flags;
1982
1983                 /* deal with zero-length terminal packet */
1984                 if (size == 0) {
1985                         if (rxhdr_flags & RXRPC_LAST_PACKET) {
1986                                 ret = rxrpc_call_flush(call);
1987                                 if (ret < 0)
1988                                         goto out;
1989                         }
1990                         break;
1991                 }
1992
1993                 /* work out how much space current packet has available */
1994                 space = call->conn->mtu_size - msg->dsize;
1995                 chunk = min(space, size);
1996
1997                 _debug("- [before] space=%Zu chunk=%Zu", space, chunk);
1998
1999                 while (!siov->iov_len)
2000                         siov++;
2001
2002                 /* if we are going to have to duplicate the data then coalesce
2003                  * it too */
2004                 if (dup_data) {
2005                         /* don't allocate more that 1 page at a time */
2006                         if (chunk > PAGE_SIZE)
2007                                 chunk = PAGE_SIZE;
2008
2009                         /* allocate a data buffer and attach to the message */
2010                         buf = kmalloc(chunk, alloc_flags);
2011                         if (unlikely(!buf)) {
2012                                 if (msg->dsize ==
2013                                     sizeof(struct rxrpc_header)) {
2014                                         /* discard an empty msg and wind back
2015                                          * the seq counter */
2016                                         rxrpc_put_message(msg);
2017                                         call->snd_nextmsg = NULL;
2018                                         call->snd_seq_count--;
2019                                 }
2020
2021                                 ret = -ENOMEM;
2022                                 goto out;
2023                         }
2024
2025                         tmp = msg->dcount++;
2026                         set_bit(tmp, &msg->dfree);
2027                         msg->data[tmp].iov_base = buf;
2028                         msg->data[tmp].iov_len = chunk;
2029                         msg->dsize += chunk;
2030                         *size_sent += chunk;
2031                         size -= chunk;
2032
2033                         /* load the buffer with data */
2034                         while (chunk > 0) {
2035                                 tmp = min(chunk, siov->iov_len);
2036                                 memcpy(buf, siov->iov_base, tmp);
2037                                 buf += tmp;
2038                                 siov->iov_base += tmp;
2039                                 siov->iov_len -= tmp;
2040                                 if (!siov->iov_len)
2041                                         siov++;
2042                                 chunk -= tmp;
2043                         }
2044                 }
2045                 else {
2046                         /* we want to attach the supplied buffers directly */
2047                         while (chunk > 0 &&
2048                                msg->dcount < RXRPC_MSG_MAX_IOCS) {
2049                                 tmp = msg->dcount++;
2050                                 msg->data[tmp].iov_base = siov->iov_base;
2051                                 msg->data[tmp].iov_len = siov->iov_len;
2052                                 msg->dsize += siov->iov_len;
2053                                 *size_sent += siov->iov_len;
2054                                 size -= siov->iov_len;
2055                                 chunk -= siov->iov_len;
2056                                 siov++;
2057                         }
2058                 }
2059
2060                 _debug("- [loaded] chunk=%Zu size=%Zu", chunk, size);
2061
2062                 /* dispatch the message when full, final or requesting ACK */
2063                 if (msg->dsize >= call->conn->mtu_size || rxhdr_flags) {
2064                         ret = rxrpc_call_flush(call);
2065                         if (ret < 0)
2066                                 goto out;
2067                 }
2068
2069         } while(size > 0);
2070
2071         ret = 0;
2072  out:
2073         _leave(" = %d (%Zd queued, %Zd rem)", ret, *size_sent, size);
2074         return ret;
2075
2076 } /* end rxrpc_call_write_data() */
2077
2078 /*****************************************************************************/
2079 /*
2080  * flush outstanding packets to the network
2081  */
2082 int rxrpc_call_flush(struct rxrpc_call *call)
2083 {
2084         struct rxrpc_message *msg;
2085         int ret = 0;
2086
2087         _enter("%p", call);
2088
2089         rxrpc_get_call(call);
2090
2091         /* if there's a packet under construction, then dispatch it now */
2092         if (call->snd_nextmsg) {
2093                 msg = call->snd_nextmsg;
2094                 call->snd_nextmsg = NULL;
2095
2096                 if (msg->hdr.flags & RXRPC_LAST_PACKET) {
2097                         msg->hdr.flags &= ~RXRPC_MORE_PACKETS;
2098                         if (call->app_call_state != RXRPC_CSTATE_CLNT_SND_ARGS)
2099                                 msg->hdr.flags |= RXRPC_REQUEST_ACK;
2100                 }
2101                 else {
2102                         msg->hdr.flags |= RXRPC_MORE_PACKETS;
2103                 }
2104
2105                 _proto("Sending DATA message { ds=%Zu dc=%u df=%02lu }",
2106                        msg->dsize, msg->dcount, msg->dfree);
2107
2108                 /* queue and adjust call state */
2109                 spin_lock(&call->lock);
2110                 list_add_tail(&msg->link, &call->acks_pendq);
2111
2112                 /* decide what to do depending on current state and if this is
2113                  * the last packet */
2114                 ret = -EINVAL;
2115                 switch (call->app_call_state) {
2116                 case RXRPC_CSTATE_SRVR_SND_REPLY:
2117                         if (msg->hdr.flags & RXRPC_LAST_PACKET) {
2118                                 call->app_call_state =
2119                                         RXRPC_CSTATE_SRVR_RCV_FINAL_ACK;
2120                                 _state(call);
2121                         }
2122                         break;
2123
2124                 case RXRPC_CSTATE_CLNT_SND_ARGS:
2125                         if (msg->hdr.flags & RXRPC_LAST_PACKET) {
2126                                 call->app_call_state =
2127                                         RXRPC_CSTATE_CLNT_RCV_REPLY;
2128                                 _state(call);
2129                         }
2130                         break;
2131
2132                 case RXRPC_CSTATE_ERROR:
2133                         ret = call->app_errno;
2134                 default:
2135                         spin_unlock(&call->lock);
2136                         goto out;
2137                 }
2138
2139                 call->acks_pend_cnt++;
2140
2141                 mod_timer(&call->acks_timeout,
2142                           __rxrpc_rtt_based_timeout(call,
2143                                                     rxrpc_call_acks_timeout));
2144
2145                 spin_unlock(&call->lock);
2146
2147                 ret = rxrpc_conn_sendmsg(call->conn, msg);
2148                 if (ret == 0)
2149                         call->pkt_snd_count++;
2150         }
2151
2152  out:
2153         rxrpc_put_call(call);
2154
2155         _leave(" = %d", ret);
2156         return ret;
2157
2158 } /* end rxrpc_call_flush() */
2159
2160 /*****************************************************************************/
2161 /*
2162  * resend NAK'd or unacknowledged packets up to the highest one specified
2163  */
2164 static void rxrpc_call_resend(struct rxrpc_call *call, rxrpc_seq_t highest)
2165 {
2166         struct rxrpc_message *msg;
2167         struct list_head *_p;
2168         rxrpc_seq_t seq = 0;
2169
2170         _enter("%p,%u", call, highest);
2171
2172         _proto("Rx Resend required");
2173
2174         /* handle too many resends */
2175         if (call->snd_resend_cnt >= rxrpc_call_max_resend) {
2176                 _debug("Aborting due to too many resends (rcv=%d)",
2177                        call->pkt_rcv_count);
2178                 rxrpc_call_abort(call,
2179                                  call->pkt_rcv_count > 0 ? -EIO : -ETIMEDOUT);
2180                 _leave("");
2181                 return;
2182         }
2183
2184         spin_lock(&call->lock);
2185         call->snd_resend_cnt++;
2186         for (;;) {
2187                 /* determine which the next packet we might need to ACK is */
2188                 if (seq <= call->acks_dftv_seq)
2189                         seq = call->acks_dftv_seq;
2190                 seq++;
2191
2192                 if (seq > highest)
2193                         break;
2194
2195                 /* look for the packet in the pending-ACK queue */
2196                 list_for_each(_p, &call->acks_pendq) {
2197                         msg = list_entry(_p, struct rxrpc_message, link);
2198                         if (msg->seq == seq)
2199                                 goto found_msg;
2200                 }
2201
2202                 panic("%s(%p,%d):"
2203                       " Inconsistent pending-ACK queue (ds=%u sc=%u sq=%u)\n",
2204                       __FUNCTION__, call, highest,
2205                       call->acks_dftv_seq, call->snd_seq_count, seq);
2206
2207         found_msg:
2208                 if (msg->state != RXRPC_MSG_SENT)
2209                         continue; /* only un-ACK'd packets */
2210
2211                 rxrpc_get_message(msg);
2212                 spin_unlock(&call->lock);
2213
2214                 /* send each message again (and ignore any errors we might
2215                  * incur) */
2216                 _proto("Resending DATA message { ds=%Zu dc=%u df=%02lu }",
2217                        msg->dsize, msg->dcount, msg->dfree);
2218
2219                 if (rxrpc_conn_sendmsg(call->conn, msg) == 0)
2220                         call->pkt_snd_count++;
2221
2222                 rxrpc_put_message(msg);
2223
2224                 spin_lock(&call->lock);
2225         }
2226
2227         /* reset the timeout */
2228         mod_timer(&call->acks_timeout,
2229                   __rxrpc_rtt_based_timeout(call, rxrpc_call_acks_timeout));
2230
2231         spin_unlock(&call->lock);
2232
2233         _leave("");
2234 } /* end rxrpc_call_resend() */
2235
2236 /*****************************************************************************/
2237 /*
2238  * handle an ICMP error being applied to a call
2239  */
2240 void rxrpc_call_handle_error(struct rxrpc_call *call, int local, int errno)
2241 {
2242         _enter("%p{%u},%d", call, ntohl(call->call_id), errno);
2243
2244         /* if this call is already aborted, then just wake up any waiters */
2245         if (call->app_call_state == RXRPC_CSTATE_ERROR) {
2246                 call->app_error_func(call);
2247         }
2248         else {
2249                 /* tell the app layer what happened */
2250                 spin_lock(&call->lock);
2251                 call->app_call_state = RXRPC_CSTATE_ERROR;
2252                 _state(call);
2253                 if (local)
2254                         call->app_err_state = RXRPC_ESTATE_LOCAL_ERROR;
2255                 else
2256                         call->app_err_state = RXRPC_ESTATE_REMOTE_ERROR;
2257                 call->app_errno         = errno;
2258                 call->app_mark          = RXRPC_APP_MARK_EOF;
2259                 call->app_read_buf      = NULL;
2260                 call->app_async_read    = 0;
2261
2262                 /* map the error */
2263                 call->app_aemap_func(call);
2264
2265                 del_timer_sync(&call->acks_timeout);
2266                 del_timer_sync(&call->rcv_timeout);
2267                 del_timer_sync(&call->ackr_dfr_timo);
2268
2269                 spin_unlock(&call->lock);
2270
2271                 call->app_error_func(call);
2272         }
2273
2274         _leave("");
2275 } /* end rxrpc_call_handle_error() */