net/9p/mux.c

   1 /*
   2  * net/9p/mux.c
   3  *
   4  * Protocol Multiplexer
   5  *
   6  *  Copyright (C) 2004 by Eric Van Hensbergen <ericvh@gmail.com>
   7  *  Copyright (C) 2004-2005 by Latchesar Ionkov <lucho@ionkov.net>
   8  *
   9  *  This program is free software; you can redistribute it and/or modify
  10  *  it under the terms of the GNU General Public License version 2
  11  *  as published by the Free Software Foundation.
  12  *
  13  *  This program is distributed in the hope that it will be useful,
  14  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  *  GNU General Public License for more details.
  17  *
  18  *  You should have received a copy of the GNU General Public License
  19  *  along with this program; if not, write to:
  20  *  Free Software Foundation
  21  *  51 Franklin Street, Fifth Floor
  22  *  Boston, MA  02111-1301  USA
  23  *
  24  */
  25
  26 #include <linux/module.h>
  27 #include <linux/errno.h>
  28 #include <linux/fs.h>
  29 #include <linux/poll.h>
  30 #include <linux/kthread.h>
  31 #include <linux/idr.h>
  32 #include <linux/mutex.h>
  33 #include <net/9p/9p.h>
  34 #include <linux/parser.h>
  35 #include <net/9p/transport.h>
  36 #include <net/9p/conn.h>
  37
  38 #define ERREQFLUSH      1
  39 #define SCHED_TIMEOUT   10
  40 #define MAXPOLLWADDR    2
  41
  42 enum {
  43         Rworksched = 1,         /* read work scheduled or running */
  44         Rpending = 2,           /* can read */
  45         Wworksched = 4,         /* write work scheduled or running */
  46         Wpending = 8,           /* can write */
  47 };
  48
  49 enum {
  50         None,
  51         Flushing,
  52         Flushed,
  53 };
  54
  55 struct p9_mux_poll_task;
  56
  57 struct p9_req {
  58         spinlock_t lock; /* protect request structure */
  59         int tag;
  60         struct p9_fcall *tcall;
  61         struct p9_fcall *rcall;
  62         int err;
  63         p9_conn_req_callback cb;
  64         void *cba;
  65         int flush;
  66         struct list_head req_list;
  67 };
  68
  69 struct p9_conn {
  70         spinlock_t lock; /* protect lock structure */
  71         struct list_head mux_list;
  72         struct p9_mux_poll_task *poll_task;
  73         int msize;
  74         unsigned char *extended;
  75         struct p9_trans *trans;
  76         struct p9_idpool *tagpool;
  77         int err;
  78         wait_queue_head_t equeue;
  79         struct list_head req_list;
  80         struct list_head unsent_req_list;
  81         struct p9_fcall *rcall;
  82         int rpos;
  83         char *rbuf;
  84         int wpos;
  85         int wsize;
  86         char *wbuf;
  87         wait_queue_t poll_wait[MAXPOLLWADDR];
  88         wait_queue_head_t *poll_waddr[MAXPOLLWADDR];
  89         poll_table pt;
  90         struct work_struct rq;
  91         struct work_struct wq;
  92         unsigned long wsched;
  93 };
  94
  95 struct p9_mux_poll_task {
  96         struct task_struct *task;
  97         struct list_head mux_list;
  98         int muxnum;
  99 };
 100
 101 struct p9_mux_rpc {
 102         struct p9_conn *m;
 103         int err;
 104         struct p9_fcall *tcall;
 105         struct p9_fcall *rcall;
 106         wait_queue_head_t wqueue;
 107 };
 108
 109 static int p9_poll_proc(void *);
 110 static void p9_read_work(struct work_struct *work);
 111 static void p9_write_work(struct work_struct *work);
 112 static void p9_pollwait(struct file *filp, wait_queue_head_t *wait_address,
 113                           poll_table * p);
 114 static u16 p9_mux_get_tag(struct p9_conn *);
 115 static void p9_mux_put_tag(struct p9_conn *, u16);
 116
 117 static DEFINE_MUTEX(p9_mux_task_lock);
 118 static struct workqueue_struct *p9_mux_wq;
 119
 120 static int p9_mux_num;
 121 static int p9_mux_poll_task_num;
 122 static struct p9_mux_poll_task p9_mux_poll_tasks[100];
 123
 124 int p9_mux_global_init(void)
 125 {
 126         int i;
 127
 128         for (i = 0; i < ARRAY_SIZE(p9_mux_poll_tasks); i++)
 129                 p9_mux_poll_tasks[i].task = NULL;
 130
 131         p9_mux_wq = create_workqueue("v9fs");
 132         if (!p9_mux_wq) {
 133                 printk(KERN_WARNING "v9fs: mux: creating workqueue failed\n");
 134                 return -ENOMEM;
 135         }
 136
 137         return 0;
 138 }
 139
 140 void p9_mux_global_exit(void)
 141 {
 142         destroy_workqueue(p9_mux_wq);
 143 }
 144
 145 /**
 146  * p9_mux_calc_poll_procs - calculates the number of polling procs
 147  * based on the number of mounted v9fs filesystems.
 148  *
 149  * The current implementation returns sqrt of the number of mounts.
 150  */
 151 static int p9_mux_calc_poll_procs(int muxnum)
 152 {
 153         int n;
 154
 155         if (p9_mux_poll_task_num)
 156                 n = muxnum / p9_mux_poll_task_num +
 157                     (muxnum % p9_mux_poll_task_num ? 1 : 0);
 158         else
 159                 n = 1;
 160
 161         if (n > ARRAY_SIZE(p9_mux_poll_tasks))
 162                 n = ARRAY_SIZE(p9_mux_poll_tasks);
 163
 164         return n;
 165 }
 166
 167 static int p9_mux_poll_start(struct p9_conn *m)
 168 {
 169         int i, n;
 170         struct p9_mux_poll_task *vpt, *vptlast;
 171         struct task_struct *pproc;
 172
 173         P9_DPRINTK(P9_DEBUG_MUX, "mux %p muxnum %d procnum %d\n", m, p9_mux_num,
 174                 p9_mux_poll_task_num);
 175         mutex_lock(&p9_mux_task_lock);
 176
 177         n = p9_mux_calc_poll_procs(p9_mux_num + 1);
 178         if (n > p9_mux_poll_task_num) {
 179                 for (i = 0; i < ARRAY_SIZE(p9_mux_poll_tasks); i++) {
 180                         if (p9_mux_poll_tasks[i].task == NULL) {
 181                                 vpt = &p9_mux_poll_tasks[i];
 182                                 P9_DPRINTK(P9_DEBUG_MUX, "create proc %p\n",
 183                                                                         vpt);
 184                                 pproc = kthread_create(p9_poll_proc, vpt,
 185                                                                 "v9fs-poll");
 186
 187                                 if (!IS_ERR(pproc)) {
 188                                         vpt->task = pproc;
 189                                         INIT_LIST_HEAD(&vpt->mux_list);
 190                                         vpt->muxnum = 0;
 191                                         p9_mux_poll_task_num++;
 192                                         wake_up_process(vpt->task);
 193                                 }
 194                                 break;
 195                         }
 196                 }
 197
 198                 if (i >= ARRAY_SIZE(p9_mux_poll_tasks))
 199                         P9_DPRINTK(P9_DEBUG_ERROR,
 200                                         "warning: no free poll slots\n");
 201         }
 202
 203         n = (p9_mux_num + 1) / p9_mux_poll_task_num +
 204             ((p9_mux_num + 1) % p9_mux_poll_task_num ? 1 : 0);
 205
 206         vptlast = NULL;
 207         for (i = 0; i < ARRAY_SIZE(p9_mux_poll_tasks); i++) {
 208                 vpt = &p9_mux_poll_tasks[i];
 209                 if (vpt->task != NULL) {
 210                         vptlast = vpt;
 211                         if (vpt->muxnum < n) {
 212                                 P9_DPRINTK(P9_DEBUG_MUX, "put in proc %d\n", i);
 213                                 list_add(&m->mux_list, &vpt->mux_list);
 214                                 vpt->muxnum++;
 215                                 m->poll_task = vpt;
 216                                 memset(&m->poll_waddr, 0,
 217                                                         sizeof(m->poll_waddr));
 218                                 init_poll_funcptr(&m->pt, p9_pollwait);
 219                                 break;
 220                         }
 221                 }
 222         }
 223
 224         if (i >= ARRAY_SIZE(p9_mux_poll_tasks)) {
 225                 if (vptlast == NULL) {
 226                         mutex_unlock(&p9_mux_task_lock);
 227                         return -ENOMEM;
 228                 }
 229
 230                 P9_DPRINTK(P9_DEBUG_MUX, "put in proc %d\n", i);
 231                 list_add(&m->mux_list, &vptlast->mux_list);
 232                 vptlast->muxnum++;
 233                 m->poll_task = vptlast;
 234                 memset(&m->poll_waddr, 0, sizeof(m->poll_waddr));
 235                 init_poll_funcptr(&m->pt, p9_pollwait);
 236         }
 237
 238         p9_mux_num++;
 239         mutex_unlock(&p9_mux_task_lock);
 240
 241         return 0;
 242 }
 243
 244 static void p9_mux_poll_stop(struct p9_conn *m)
 245 {
 246         int i;
 247         struct p9_mux_poll_task *vpt;
 248
 249         mutex_lock(&p9_mux_task_lock);
 250         vpt = m->poll_task;
 251         list_del(&m->mux_list);
 252         for (i = 0; i < ARRAY_SIZE(m->poll_waddr); i++) {
 253                 if (m->poll_waddr[i] != NULL) {
 254                         remove_wait_queue(m->poll_waddr[i], &m->poll_wait[i]);
 255                         m->poll_waddr[i] = NULL;
 256                 }
 257         }
 258         vpt->muxnum--;
 259         if (!vpt->muxnum) {
 260                 P9_DPRINTK(P9_DEBUG_MUX, "destroy proc %p\n", vpt);
 261                 kthread_stop(vpt->task);
 262                 vpt->task = NULL;
 263                 p9_mux_poll_task_num--;
 264         }
 265         p9_mux_num--;
 266         mutex_unlock(&p9_mux_task_lock);
 267 }
 268
 269 /**
 270  * p9_conn_create - allocate and initialize the per-session mux data
 271  * Creates the polling task if this is the first session.
 272  *
 273  * @trans - transport structure
 274  * @msize - maximum message size
 275  * @extended - pointer to the extended flag
 276  */
 277 struct p9_conn *p9_conn_create(struct p9_trans *trans, int msize,
 278                                     unsigned char *extended)
 279 {
 280         int i, n;
 281         struct p9_conn *m, *mtmp;
 282
 283         P9_DPRINTK(P9_DEBUG_MUX, "transport %p msize %d\n", trans, msize);
 284         m = kmalloc(sizeof(struct p9_conn), GFP_KERNEL);
 285         if (!m)
 286                 return ERR_PTR(-ENOMEM);
 287
 288         spin_lock_init(&m->lock);
 289         INIT_LIST_HEAD(&m->mux_list);
 290         m->msize = msize;
 291         m->extended = extended;
 292         m->trans = trans;
 293         m->tagpool = p9_idpool_create();
 294         if (IS_ERR(m->tagpool)) {
 295                 mtmp = ERR_PTR(-ENOMEM);
 296                 kfree(m);
 297                 return mtmp;
 298         }
 299
 300         m->err = 0;
 301         init_waitqueue_head(&m->equeue);
 302         INIT_LIST_HEAD(&m->req_list);
 303         INIT_LIST_HEAD(&m->unsent_req_list);
 304         m->rcall = NULL;
 305         m->rpos = 0;
 306         m->rbuf = NULL;
 307         m->wpos = m->wsize = 0;
 308         m->wbuf = NULL;
 309         INIT_WORK(&m->rq, p9_read_work);
 310         INIT_WORK(&m->wq, p9_write_work);
 311         m->wsched = 0;
 312         memset(&m->poll_waddr, 0, sizeof(m->poll_waddr));
 313         m->poll_task = NULL;
 314         n = p9_mux_poll_start(m);
 315         if (n) {
 316                 kfree(m);
 317                 return ERR_PTR(n);
 318         }
 319
 320         n = trans->poll(trans, &m->pt);
 321         if (n & POLLIN) {
 322                 P9_DPRINTK(P9_DEBUG_MUX, "mux %p can read\n", m);
 323                 set_bit(Rpending, &m->wsched);
 324         }
 325
 326         if (n & POLLOUT) {
 327                 P9_DPRINTK(P9_DEBUG_MUX, "mux %p can write\n", m);
 328                 set_bit(Wpending, &m->wsched);
 329         }
 330
 331         for (i = 0; i < ARRAY_SIZE(m->poll_waddr); i++) {
 332                 if (IS_ERR(m->poll_waddr[i])) {
 333                         p9_mux_poll_stop(m);
 334                         mtmp = (void *)m->poll_waddr;   /* the error code */
 335                         kfree(m);
 336                         m = mtmp;
 337                         break;
 338                 }
 339         }
 340
 341         return m;
 342 }
 343 EXPORT_SYMBOL(p9_conn_create);
 344
 345 /**
 346  * p9_mux_destroy - cancels all pending requests and frees mux resources
 347  */
 348 void p9_conn_destroy(struct p9_conn *m)
 349 {
 350         P9_DPRINTK(P9_DEBUG_MUX, "mux %p prev %p next %p\n", m,
 351                 m->mux_list.prev, m->mux_list.next);
 352         p9_conn_cancel(m, -ECONNRESET);
 353
 354         if (!list_empty(&m->req_list)) {
 355                 /* wait until all processes waiting on this session exit */
 356                 P9_DPRINTK(P9_DEBUG_MUX,
 357                         "mux %p waiting for empty request queue\n", m);
 358                 wait_event_timeout(m->equeue, (list_empty(&m->req_list)), 5000);
 359                 P9_DPRINTK(P9_DEBUG_MUX, "mux %p request queue empty: %d\n", m,
 360                         list_empty(&m->req_list));
 361         }
 362
 363         p9_mux_poll_stop(m);
 364         m->trans = NULL;
 365         p9_idpool_destroy(m->tagpool);
 366         kfree(m);
 367 }
 368 EXPORT_SYMBOL(p9_conn_destroy);
 369
 370 /**
 371  * p9_pollwait - called by files poll operation to add v9fs-poll task
 372  *      to files wait queue
 373  */
 374 static void
 375 p9_pollwait(struct file *filp, wait_queue_head_t *wait_address,
 376               poll_table * p)
 377 {
 378         int i;
 379         struct p9_conn *m;
 380
 381         m = container_of(p, struct p9_conn, pt);
 382         for (i = 0; i < ARRAY_SIZE(m->poll_waddr); i++)
 383                 if (m->poll_waddr[i] == NULL)
 384                         break;
 385
 386         if (i >= ARRAY_SIZE(m->poll_waddr)) {
 387                 P9_DPRINTK(P9_DEBUG_ERROR, "not enough wait_address slots\n");
 388                 return;
 389         }
 390
 391         m->poll_waddr[i] = wait_address;
 392
 393         if (!wait_address) {
 394                 P9_DPRINTK(P9_DEBUG_ERROR, "no wait_address\n");
 395                 m->poll_waddr[i] = ERR_PTR(-EIO);
 396                 return;
 397         }
 398
 399         init_waitqueue_entry(&m->poll_wait[i], m->poll_task->task);
 400         add_wait_queue(wait_address, &m->poll_wait[i]);
 401 }
 402
 403 /**
 404  * p9_poll_mux - polls a mux and schedules read or write works if necessary
 405  */
 406 static void p9_poll_mux(struct p9_conn *m)
 407 {
 408         int n;
 409
 410         if (m->err < 0)
 411                 return;
 412
 413         n = m->trans->poll(m->trans, NULL);
 414         if (n < 0 || n & (POLLERR | POLLHUP | POLLNVAL)) {
 415                 P9_DPRINTK(P9_DEBUG_MUX, "error mux %p err %d\n", m, n);
 416                 if (n >= 0)
 417                         n = -ECONNRESET;
 418                 p9_conn_cancel(m, n);
 419         }
 420
 421         if (n & POLLIN) {
 422                 set_bit(Rpending, &m->wsched);
 423                 P9_DPRINTK(P9_DEBUG_MUX, "mux %p can read\n", m);
 424                 if (!test_and_set_bit(Rworksched, &m->wsched)) {
 425                         P9_DPRINTK(P9_DEBUG_MUX, "schedule read work %p\n", m);
 426                         queue_work(p9_mux_wq, &m->rq);
 427                 }
 428         }
 429
 430         if (n & POLLOUT) {
 431                 set_bit(Wpending, &m->wsched);
 432                 P9_DPRINTK(P9_DEBUG_MUX, "mux %p can write\n", m);
 433                 if ((m->wsize || !list_empty(&m->unsent_req_list))
 434                     && !test_and_set_bit(Wworksched, &m->wsched)) {
 435                         P9_DPRINTK(P9_DEBUG_MUX, "schedule write work %p\n", m);
 436                         queue_work(p9_mux_wq, &m->wq);
 437                 }
 438         }
 439 }
 440
 441 /**
 442  * p9_poll_proc - polls all v9fs transports for new events and queues
 443  *      the appropriate work to the work queue
 444  */
 445 static int p9_poll_proc(void *a)
 446 {
 447         struct p9_conn *m, *mtmp;
 448         struct p9_mux_poll_task *vpt;
 449
 450         vpt = a;
 451         P9_DPRINTK(P9_DEBUG_MUX, "start %p %p\n", current, vpt);
 452         while (!kthread_should_stop()) {
 453                 set_current_state(TASK_INTERRUPTIBLE);
 454
 455                 list_for_each_entry_safe(m, mtmp, &vpt->mux_list, mux_list) {
 456                         p9_poll_mux(m);
 457                 }
 458
 459                 P9_DPRINTK(P9_DEBUG_MUX, "sleeping...\n");
 460                 schedule_timeout(SCHED_TIMEOUT * HZ);
 461         }
 462
 463         __set_current_state(TASK_RUNNING);
 464         P9_DPRINTK(P9_DEBUG_MUX, "finish\n");
 465         return 0;
 466 }
 467
 468 /**
 469  * p9_write_work - called when a transport can send some data
 470  */
 471 static void p9_write_work(struct work_struct *work)
 472 {
 473         int n, err;
 474         struct p9_conn *m;
 475         struct p9_req *req;
 476
 477         m = container_of(work, struct p9_conn, wq);
 478
 479         if (m->err < 0) {
 480                 clear_bit(Wworksched, &m->wsched);
 481                 return;
 482         }
 483
 484         if (!m->wsize) {
 485                 if (list_empty(&m->unsent_req_list)) {
 486                         clear_bit(Wworksched, &m->wsched);
 487                         return;
 488                 }
 489
 490                 spin_lock(&m->lock);
 491 again:
 492                 req = list_entry(m->unsent_req_list.next, struct p9_req,
 493                                req_list);
 494                 list_move_tail(&req->req_list, &m->req_list);
 495                 if (req->err == ERREQFLUSH)
 496                         goto again;
 497
 498                 m->wbuf = req->tcall->sdata;
 499                 m->wsize = req->tcall->size;
 500                 m->wpos = 0;
 501                 spin_unlock(&m->lock);
 502         }
 503
 504         P9_DPRINTK(P9_DEBUG_MUX, "mux %p pos %d size %d\n", m, m->wpos,
 505                                                                 m->wsize);
 506         clear_bit(Wpending, &m->wsched);
 507         err = m->trans->write(m->trans, m->wbuf + m->wpos, m->wsize - m->wpos);
 508         P9_DPRINTK(P9_DEBUG_MUX, "mux %p sent %d bytes\n", m, err);
 509         if (err == -EAGAIN) {
 510                 clear_bit(Wworksched, &m->wsched);
 511                 return;
 512         }
 513
 514         if (err < 0)
 515                 goto error;
 516         else if (err == 0) {
 517                 err = -EREMOTEIO;
 518                 goto error;
 519         }
 520
 521         m->wpos += err;
 522         if (m->wpos == m->wsize)
 523                 m->wpos = m->wsize = 0;
 524
 525         if (m->wsize == 0 && !list_empty(&m->unsent_req_list)) {
 526                 if (test_and_clear_bit(Wpending, &m->wsched))
 527                         n = POLLOUT;
 528                 else
 529                         n = m->trans->poll(m->trans, NULL);
 530
 531                 if (n & POLLOUT) {
 532                         P9_DPRINTK(P9_DEBUG_MUX, "schedule write work %p\n", m);
 533                         queue_work(p9_mux_wq, &m->wq);
 534                 } else
 535                         clear_bit(Wworksched, &m->wsched);
 536         } else
 537                 clear_bit(Wworksched, &m->wsched);
 538
 539         return;
 540
 541 error:
 542         p9_conn_cancel(m, err);
 543         clear_bit(Wworksched, &m->wsched);
 544 }
 545
 546 static void process_request(struct p9_conn *m, struct p9_req *req)
 547 {
 548         int ecode;
 549         struct p9_str *ename;
 550
 551         if (!req->err && req->rcall->id == P9_RERROR) {
 552                 ecode = req->rcall->params.rerror.errno;
 553                 ename = &req->rcall->params.rerror.error;
 554
 555                 P9_DPRINTK(P9_DEBUG_MUX, "Rerror %.*s\n", ename->len,
 556                                                                 ename->str);
 557
 558                 if (*m->extended)
 559                         req->err = -ecode;
 560
 561                 if (!req->err) {
 562                         req->err = p9_errstr2errno(ename->str, ename->len);
 563
 564                         if (!req->err) {        /* string match failed */
 565                                 PRINT_FCALL_ERROR("unknown error", req->rcall);
 566                         }
 567
 568                         if (!req->err)
 569                                 req->err = -ESERVERFAULT;
 570                 }
 571         } else if (req->tcall && req->rcall->id != req->tcall->id + 1) {
 572                 P9_DPRINTK(P9_DEBUG_ERROR,
 573                                 "fcall mismatch: expected %d, got %d\n",
 574                                 req->tcall->id + 1, req->rcall->id);
 575                 if (!req->err)
 576                         req->err = -EIO;
 577         }
 578 }
 579
 580 /**
 581  * p9_read_work - called when there is some data to be read from a transport
 582  */
 583 static void p9_read_work(struct work_struct *work)
 584 {
 585         int n, err;
 586         struct p9_conn *m;
 587         struct p9_req *req, *rptr, *rreq;
 588         struct p9_fcall *rcall;
 589         char *rbuf;
 590
 591         m = container_of(work, struct p9_conn, rq);
 592
 593         if (m->err < 0)
 594                 return;
 595
 596         rcall = NULL;
 597         P9_DPRINTK(P9_DEBUG_MUX, "start mux %p pos %d\n", m, m->rpos);
 598
 599         if (!m->rcall) {
 600                 m->rcall =
 601                     kmalloc(sizeof(struct p9_fcall) + m->msize, GFP_KERNEL);
 602                 if (!m->rcall) {
 603                         err = -ENOMEM;
 604                         goto error;
 605                 }
 606
 607                 m->rbuf = (char *)m->rcall + sizeof(struct p9_fcall);
 608                 m->rpos = 0;
 609         }
 610
 611         clear_bit(Rpending, &m->wsched);
 612         err = m->trans->read(m->trans, m->rbuf + m->rpos, m->msize - m->rpos);
 613         P9_DPRINTK(P9_DEBUG_MUX, "mux %p got %d bytes\n", m, err);
 614         if (err == -EAGAIN) {
 615                 clear_bit(Rworksched, &m->wsched);
 616                 return;
 617         }
 618
 619         if (err <= 0)
 620                 goto error;
 621
 622         m->rpos += err;
 623         while (m->rpos > 4) {
 624                 n = le32_to_cpu(*(__le32 *) m->rbuf);
 625                 if (n >= m->msize) {
 626                         P9_DPRINTK(P9_DEBUG_ERROR,
 627                                 "requested packet size too big: %d\n", n);
 628                         err = -EIO;
 629                         goto error;
 630                 }
 631
 632                 if (m->rpos < n)
 633                         break;
 634
 635                 err =
 636                     p9_deserialize_fcall(m->rbuf, n, m->rcall, *m->extended);
 637                 if (err < 0) {
 638                         goto error;
 639                 }
 640
 641 #ifdef CONFIG_NET_9P_DEBUG
 642                 if ((p9_debug_level&P9_DEBUG_FCALL) == P9_DEBUG_FCALL) {
 643                         char buf[150];
 644
 645                         p9_printfcall(buf, sizeof(buf), m->rcall,
 646                                 *m->extended);
 647                         printk(KERN_NOTICE ">>> %p %s\n", m, buf);
 648                 }
 649 #endif
 650
 651                 rcall = m->rcall;
 652                 rbuf = m->rbuf;
 653                 if (m->rpos > n) {
 654                         m->rcall = kmalloc(sizeof(struct p9_fcall) + m->msize,
 655                                            GFP_KERNEL);
 656                         if (!m->rcall) {
 657                                 err = -ENOMEM;
 658                                 goto error;
 659                         }
 660
 661                         m->rbuf = (char *)m->rcall + sizeof(struct p9_fcall);
 662                         memmove(m->rbuf, rbuf + n, m->rpos - n);
 663                         m->rpos -= n;
 664                 } else {
 665                         m->rcall = NULL;
 666                         m->rbuf = NULL;
 667                         m->rpos = 0;
 668                 }
 669
 670                 P9_DPRINTK(P9_DEBUG_MUX, "mux %p fcall id %d tag %d\n", m,
 671                                                         rcall->id, rcall->tag);
 672
 673                 req = NULL;
 674                 spin_lock(&m->lock);
 675                 list_for_each_entry_safe(rreq, rptr, &m->req_list, req_list) {
 676                         if (rreq->tag == rcall->tag) {
 677                                 req = rreq;
 678                                 if (req->flush != Flushing)
 679                                         list_del(&req->req_list);
 680                                 break;
 681                         }
 682                 }
 683                 spin_unlock(&m->lock);
 684
 685                 if (req) {
 686                         req->rcall = rcall;
 687                         process_request(m, req);
 688
 689                         if (req->flush != Flushing) {
 690                                 if (req->cb)
 691                                         (*req->cb) (req, req->cba);
 692                                 else
 693                                         kfree(req->rcall);
 694
 695                                 wake_up(&m->equeue);
 696                         }
 697                 } else {
 698                         if (err >= 0 && rcall->id != P9_RFLUSH)
 699                                 P9_DPRINTK(P9_DEBUG_ERROR,
 700                                   "unexpected response mux %p id %d tag %d\n",
 701                                   m, rcall->id, rcall->tag);
 702                         kfree(rcall);
 703                 }
 704         }
 705
 706         if (!list_empty(&m->req_list)) {
 707                 if (test_and_clear_bit(Rpending, &m->wsched))
 708                         n = POLLIN;
 709                 else
 710                         n = m->trans->poll(m->trans, NULL);
 711
 712                 if (n & POLLIN) {
 713                         P9_DPRINTK(P9_DEBUG_MUX, "schedule read work %p\n", m);
 714                         queue_work(p9_mux_wq, &m->rq);
 715                 } else
 716                         clear_bit(Rworksched, &m->wsched);
 717         } else
 718                 clear_bit(Rworksched, &m->wsched);
 719
 720         return;
 721
 722 error:
 723         p9_conn_cancel(m, err);
 724         clear_bit(Rworksched, &m->wsched);
 725 }
 726
 727 /**
 728  * p9_send_request - send 9P request
 729  * The function can sleep until the request is scheduled for sending.
 730  * The function can be interrupted. Return from the function is not
 731  * a guarantee that the request is sent successfully. Can return errors
 732  * that can be retrieved by PTR_ERR macros.
 733  *
 734  * @m: mux data
 735  * @tc: request to be sent
 736  * @cb: callback function to call when response is received
 737  * @cba: parameter to pass to the callback function
 738  */
 739 static struct p9_req *p9_send_request(struct p9_conn *m,
 740                                           struct p9_fcall *tc,
 741                                           p9_conn_req_callback cb, void *cba)
 742 {
 743         int n;
 744         struct p9_req *req;
 745
 746         P9_DPRINTK(P9_DEBUG_MUX, "mux %p task %p tcall %p id %d\n", m, current,
 747                 tc, tc->id);
 748         if (m->err < 0)
 749                 return ERR_PTR(m->err);
 750
 751         req = kmalloc(sizeof(struct p9_req), GFP_KERNEL);
 752         if (!req)
 753                 return ERR_PTR(-ENOMEM);
 754
 755         if (tc->id == P9_TVERSION)
 756                 n = P9_NOTAG;
 757         else
 758                 n = p9_mux_get_tag(m);
 759
 760         if (n < 0)
 761                 return ERR_PTR(-ENOMEM);
 762
 763         p9_set_tag(tc, n);
 764
 765 #ifdef CONFIG_NET_9P_DEBUG
 766         if ((p9_debug_level&P9_DEBUG_FCALL) == P9_DEBUG_FCALL) {
 767                 char buf[150];
 768
 769                 p9_printfcall(buf, sizeof(buf), tc, *m->extended);
 770                 printk(KERN_NOTICE "<<< %p %s\n", m, buf);
 771         }
 772 #endif
 773
 774         spin_lock_init(&req->lock);
 775         req->tag = n;
 776         req->tcall = tc;
 777         req->rcall = NULL;
 778         req->err = 0;
 779         req->cb = cb;
 780         req->cba = cba;
 781         req->flush = None;
 782
 783         spin_lock(&m->lock);
 784         list_add_tail(&req->req_list, &m->unsent_req_list);
 785         spin_unlock(&m->lock);
 786
 787         if (test_and_clear_bit(Wpending, &m->wsched))
 788                 n = POLLOUT;
 789         else
 790                 n = m->trans->poll(m->trans, NULL);
 791
 792         if (n & POLLOUT && !test_and_set_bit(Wworksched, &m->wsched))
 793                 queue_work(p9_mux_wq, &m->wq);
 794
 795         return req;
 796 }
 797
 798 static void p9_mux_free_request(struct p9_conn *m, struct p9_req *req)
 799 {
 800         p9_mux_put_tag(m, req->tag);
 801         kfree(req);
 802 }
 803
 804 static void p9_mux_flush_cb(struct p9_req *freq, void *a)
 805 {
 806         p9_conn_req_callback cb;
 807         int tag;
 808         struct p9_conn *m;
 809         struct p9_req *req, *rreq, *rptr;
 810
 811         m = a;
 812         P9_DPRINTK(P9_DEBUG_MUX, "mux %p tc %p rc %p err %d oldtag %d\n", m,
 813                 freq->tcall, freq->rcall, freq->err,
 814                 freq->tcall->params.tflush.oldtag);
 815
 816         spin_lock(&m->lock);
 817         cb = NULL;
 818         tag = freq->tcall->params.tflush.oldtag;
 819         req = NULL;
 820         list_for_each_entry_safe(rreq, rptr, &m->req_list, req_list) {
 821                 if (rreq->tag == tag) {
 822                         req = rreq;
 823                         list_del(&req->req_list);
 824                         break;
 825                 }
 826         }
 827         spin_unlock(&m->lock);
 828
 829         if (req) {
 830                 spin_lock(&req->lock);
 831                 req->flush = Flushed;
 832                 spin_unlock(&req->lock);
 833
 834                 if (req->cb)
 835                         (*req->cb) (req, req->cba);
 836                 else
 837                         kfree(req->rcall);
 838
 839                 wake_up(&m->equeue);
 840         }
 841
 842         kfree(freq->tcall);
 843         kfree(freq->rcall);
 844         p9_mux_free_request(m, freq);
 845 }
 846
 847 static int
 848 p9_mux_flush_request(struct p9_conn *m, struct p9_req *req)
 849 {
 850         struct p9_fcall *fc;
 851         struct p9_req *rreq, *rptr;
 852
 853         P9_DPRINTK(P9_DEBUG_MUX, "mux %p req %p tag %d\n", m, req, req->tag);
 854
 855         /* if a response was received for a request, do nothing */
 856         spin_lock(&req->lock);
 857         if (req->rcall || req->err) {
 858                 spin_unlock(&req->lock);
 859                 P9_DPRINTK(P9_DEBUG_MUX,
 860                         "mux %p req %p response already received\n", m, req);
 861                 return 0;
 862         }
 863
 864         req->flush = Flushing;
 865         spin_unlock(&req->lock);
 866
 867         spin_lock(&m->lock);
 868         /* if the request is not sent yet, just remove it from the list */
 869         list_for_each_entry_safe(rreq, rptr, &m->unsent_req_list, req_list) {
 870                 if (rreq->tag == req->tag) {
 871                         P9_DPRINTK(P9_DEBUG_MUX,
 872                            "mux %p req %p request is not sent yet\n", m, req);
 873                         list_del(&rreq->req_list);
 874                         req->flush = Flushed;
 875                         spin_unlock(&m->lock);
 876                         if (req->cb)
 877                                 (*req->cb) (req, req->cba);
 878                         return 0;
 879                 }
 880         }
 881         spin_unlock(&m->lock);
 882
 883         clear_thread_flag(TIF_SIGPENDING);
 884         fc = p9_create_tflush(req->tag);
 885         p9_send_request(m, fc, p9_mux_flush_cb, m);
 886         return 1;
 887 }
 888
 889 static void
 890 p9_conn_rpc_cb(struct p9_req *req, void *a)
 891 {
 892         struct p9_mux_rpc *r;
 893
 894         P9_DPRINTK(P9_DEBUG_MUX, "req %p r %p\n", req, a);
 895         r = a;
 896         r->rcall = req->rcall;
 897         r->err = req->err;
 898
 899         if (req->flush != None && !req->err)
 900                 r->err = -ERESTARTSYS;
 901
 902         wake_up(&r->wqueue);
 903 }
 904
 905 /**
 906  * p9_mux_rpc - sends 9P request and waits until a response is available.
 907  *      The function can be interrupted.
 908  * @m: mux data
 909  * @tc: request to be sent
 910  * @rc: pointer where a pointer to the response is stored
 911  */
 912 int
 913 p9_conn_rpc(struct p9_conn *m, struct p9_fcall *tc,
 914              struct p9_fcall **rc)
 915 {
 916         int err, sigpending;
 917         unsigned long flags;
 918         struct p9_req *req;
 919         struct p9_mux_rpc r;
 920
 921         r.err = 0;
 922         r.tcall = tc;
 923         r.rcall = NULL;
 924         r.m = m;
 925         init_waitqueue_head(&r.wqueue);
 926
 927         if (rc)
 928                 *rc = NULL;
 929
 930         sigpending = 0;
 931         if (signal_pending(current)) {
 932                 sigpending = 1;
 933                 clear_thread_flag(TIF_SIGPENDING);
 934         }
 935
 936         req = p9_send_request(m, tc, p9_conn_rpc_cb, &r);
 937         if (IS_ERR(req)) {
 938                 err = PTR_ERR(req);
 939                 P9_DPRINTK(P9_DEBUG_MUX, "error %d\n", err);
 940                 return err;
 941         }
 942
 943         err = wait_event_interruptible(r.wqueue, r.rcall != NULL || r.err < 0);
 944         if (r.err < 0)
 945                 err = r.err;
 946
 947         if (err == -ERESTARTSYS && m->trans->status == Connected
 948                                                         && m->err == 0) {
 949                 if (p9_mux_flush_request(m, req)) {
 950                         /* wait until we get response of the flush message */
 951                         do {
 952                                 clear_thread_flag(TIF_SIGPENDING);
 953                                 err = wait_event_interruptible(r.wqueue,
 954                                         r.rcall || r.err);
 955                         } while (!r.rcall && !r.err && err == -ERESTARTSYS &&
 956                                 m->trans->status == Connected && !m->err);
 957
 958                         err = -ERESTARTSYS;
 959                 }
 960                 sigpending = 1;
 961         }
 962
 963         if (sigpending) {
 964                 spin_lock_irqsave(&current->sighand->siglock, flags);
 965                 recalc_sigpending();
 966                 spin_unlock_irqrestore(&current->sighand->siglock, flags);
 967         }
 968
 969         if (rc)
 970                 *rc = r.rcall;
 971         else
 972                 kfree(r.rcall);
 973
 974         p9_mux_free_request(m, req);
 975         if (err > 0)
 976                 err = -EIO;
 977
 978         return err;
 979 }
 980 EXPORT_SYMBOL(p9_conn_rpc);
 981
 982 #ifdef P9_NONBLOCK
 983 /**
 984  * p9_conn_rpcnb - sends 9P request without waiting for response.
 985  * @m: mux data
 986  * @tc: request to be sent
 987  * @cb: callback function to be called when response arrives
 988  * @cba: value to pass to the callback function
 989  */
 990 int p9_conn_rpcnb(struct p9_conn *m, struct p9_fcall *tc,
 991                    p9_conn_req_callback cb, void *a)
 992 {
 993         int err;
 994         struct p9_req *req;
 995
 996         req = p9_send_request(m, tc, cb, a);
 997         if (IS_ERR(req)) {
 998                 err = PTR_ERR(req);
 999                 P9_DPRINTK(P9_DEBUG_MUX, "error %d\n", err);
1000                 return PTR_ERR(req);
1001         }
1002
1003         P9_DPRINTK(P9_DEBUG_MUX, "mux %p tc %p tag %d\n", m, tc, req->tag);
1004         return 0;
1005 }
1006 EXPORT_SYMBOL(p9_conn_rpcnb);
1007 #endif /* P9_NONBLOCK */
1008
1009 /**
1010  * p9_conn_cancel - cancel all pending requests with error
1011  * @m: mux data
1012  * @err: error code
1013  */
1014 void p9_conn_cancel(struct p9_conn *m, int err)
1015 {
1016         struct p9_req *req, *rtmp;
1017         LIST_HEAD(cancel_list);
1018
1019         P9_DPRINTK(P9_DEBUG_ERROR, "mux %p err %d\n", m, err);
1020         m->err = err;
1021         spin_lock(&m->lock);
1022         list_for_each_entry_safe(req, rtmp, &m->req_list, req_list) {
1023                 list_move(&req->req_list, &cancel_list);
1024         }
1025         list_for_each_entry_safe(req, rtmp, &m->unsent_req_list, req_list) {
1026                 list_move(&req->req_list, &cancel_list);
1027         }
1028         spin_unlock(&m->lock);
1029
1030         list_for_each_entry_safe(req, rtmp, &cancel_list, req_list) {
1031                 list_del(&req->req_list);
1032                 if (!req->err)
1033                         req->err = err;
1034
1035                 if (req->cb)
1036                         (*req->cb) (req, req->cba);
1037                 else
1038                         kfree(req->rcall);
1039         }
1040
1041         wake_up(&m->equeue);
1042 }
1043 EXPORT_SYMBOL(p9_conn_cancel);
1044
1045 static u16 p9_mux_get_tag(struct p9_conn *m)
1046 {
1047         int tag;
1048
1049         tag = p9_idpool_get(m->tagpool);
1050         if (tag < 0)
1051                 return P9_NOTAG;
1052         else
1053                 return (u16) tag;
1054 }
1055
1056 static void p9_mux_put_tag(struct p9_conn *m, u16 tag)
1057 {
1058         if (tag != P9_NOTAG && p9_idpool_check(tag, m->tagpool))
1059                 p9_idpool_put(tag, m->tagpool);
1060 }