drivers/staging/dst/state.c

   1 /*
   2  * 2007+ Copyright (c) Evgeniy Polyakov <zbr@ioremap.net>
   3  * All rights reserved.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  */
  15
  16 #include <linux/buffer_head.h>
  17 #include <linux/blkdev.h>
  18 #include <linux/bio.h>
  19 #include <linux/connector.h>
  20 #include <linux/dst.h>
  21 #include <linux/device.h>
  22 #include <linux/in.h>
  23 #include <linux/in6.h>
  24 #include <linux/socket.h>
  25 #include <linux/slab.h>
  26
  27 #include <net/sock.h>
  28
  29 /*
  30  * Polling machinery.
  31  */
  32
  33 struct dst_poll_helper {
  34         poll_table              pt;
  35         struct dst_state        *st;
  36 };
  37
  38 static int dst_queue_wake(wait_queue_t *wait, unsigned mode,
  39                 int sync, void *key)
  40 {
  41         struct dst_state *st = container_of(wait, struct dst_state, wait);
  42
  43         wake_up(&st->thread_wait);
  44         return 1;
  45 }
  46
  47 static void dst_queue_func(struct file *file, wait_queue_head_t *whead,
  48                                  poll_table *pt)
  49 {
  50         struct dst_state *st = container_of(pt, struct dst_poll_helper, pt)->st;
  51
  52         st->whead = whead;
  53         init_waitqueue_func_entry(&st->wait, dst_queue_wake);
  54         add_wait_queue(whead, &st->wait);
  55 }
  56
  57 void dst_poll_exit(struct dst_state *st)
  58 {
  59         if (st->whead) {
  60                 remove_wait_queue(st->whead, &st->wait);
  61                 st->whead = NULL;
  62         }
  63 }
  64
  65 int dst_poll_init(struct dst_state *st)
  66 {
  67         struct dst_poll_helper ph;
  68
  69         ph.st = st;
  70         init_poll_funcptr(&ph.pt, &dst_queue_func);
  71
  72         st->socket->ops->poll(NULL, st->socket, &ph.pt);
  73         return 0;
  74 }
  75
  76 /*
  77  * Header receiving function - may block.
  78  */
  79 static int dst_data_recv_header(struct socket *sock,
  80                 void *data, unsigned int size, int block)
  81 {
  82         struct msghdr msg;
  83         struct kvec iov;
  84         int err;
  85
  86         iov.iov_base = data;
  87         iov.iov_len = size;
  88
  89         msg.msg_iov = (struct iovec *)&iov;
  90         msg.msg_iovlen = 1;
  91         msg.msg_name = NULL;
  92         msg.msg_namelen = 0;
  93         msg.msg_control = NULL;
  94         msg.msg_controllen = 0;
  95         msg.msg_flags = (block) ? MSG_WAITALL : MSG_DONTWAIT;
  96
  97         err = kernel_recvmsg(sock, &msg, &iov, 1, iov.iov_len,
  98                         msg.msg_flags);
  99         if (err != size)
 100                 return -1;
 101
 102         return 0;
 103 }
 104
 105 /*
 106  * Header sending function - may block.
 107  */
 108 int dst_data_send_header(struct socket *sock,
 109                 void *data, unsigned int size, int more)
 110 {
 111         struct msghdr msg;
 112         struct kvec iov;
 113         int err;
 114
 115         iov.iov_base = data;
 116         iov.iov_len = size;
 117
 118         msg.msg_iov = (struct iovec *)&iov;
 119         msg.msg_iovlen = 1;
 120         msg.msg_name = NULL;
 121         msg.msg_namelen = 0;
 122         msg.msg_control = NULL;
 123         msg.msg_controllen = 0;
 124         msg.msg_flags = MSG_WAITALL | (more ? MSG_MORE : 0);
 125
 126         err = kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
 127         if (err != size) {
 128                 dprintk("%s: size: %u, more: %d, err: %d.\n",
 129                                 __func__, size, more, err);
 130                 return -1;
 131         }
 132
 133         return 0;
 134 }
 135
 136 /*
 137  * Block autoconfiguration: request size of the storage and permissions.
 138  */
 139 static int dst_request_remote_config(struct dst_state *st)
 140 {
 141         struct dst_node *n = st->node;
 142         int err = -EINVAL;
 143         struct dst_cmd *cmd = st->data;
 144
 145         memset(cmd, 0, sizeof(struct dst_cmd));
 146         cmd->cmd = DST_CFG;
 147
 148         dst_convert_cmd(cmd);
 149
 150         err = dst_data_send_header(st->socket, cmd, sizeof(struct dst_cmd), 0);
 151         if (err)
 152                 goto out;
 153
 154         err = dst_data_recv_header(st->socket, cmd, sizeof(struct dst_cmd), 1);
 155         if (err)
 156                 goto out;
 157
 158         dst_convert_cmd(cmd);
 159
 160         if (cmd->cmd != DST_CFG) {
 161                 err = -EINVAL;
 162                 dprintk("%s: checking result: cmd: %d, size reported: %llu.\n",
 163                         __func__, cmd->cmd, cmd->sector);
 164                 goto out;
 165         }
 166
 167         if (n->size != 0)
 168                 n->size = min_t(loff_t, n->size, cmd->sector);
 169         else
 170                 n->size = cmd->sector;
 171
 172         n->info->size = n->size;
 173         st->permissions = cmd->rw;
 174
 175 out:
 176         dprintk("%s: n: %p, err: %d, size: %llu, permission: %x.\n",
 177                         __func__, n, err, n->size, st->permissions);
 178         return err;
 179 }
 180
 181 /*
 182  * Socket machinery.
 183  */
 184
 185 #define DST_DEFAULT_TIMEO       20000
 186
 187 int dst_state_socket_create(struct dst_state *st)
 188 {
 189         int err;
 190         struct socket *sock;
 191         struct dst_network_ctl *ctl = &st->ctl;
 192
 193         err = sock_create(ctl->addr.sa_family, ctl->type, ctl->proto, &sock);
 194         if (err < 0)
 195                 return err;
 196
 197         sock->sk->sk_sndtimeo = sock->sk->sk_rcvtimeo =
 198                 msecs_to_jiffies(DST_DEFAULT_TIMEO);
 199         sock->sk->sk_allocation = GFP_NOIO;
 200
 201         st->socket = st->read_socket = sock;
 202         return 0;
 203 }
 204
 205 void dst_state_socket_release(struct dst_state *st)
 206 {
 207         dprintk("%s: st: %p, socket: %p, n: %p.\n",
 208                         __func__, st, st->socket, st->node);
 209         if (st->socket) {
 210                 sock_release(st->socket);
 211                 st->socket = NULL;
 212                 st->read_socket = NULL;
 213         }
 214 }
 215
 216 void dst_dump_addr(struct socket *sk, struct sockaddr *sa, char *str)
 217 {
 218         if (sk->ops->family == AF_INET) {
 219                 struct sockaddr_in *sin = (struct sockaddr_in *)sa;
 220                 printk(KERN_INFO "%s %u.%u.%u.%u:%d.\n", str,
 221                         NIPQUAD(sin->sin_addr.s_addr), ntohs(sin->sin_port));
 222         } else if (sk->ops->family == AF_INET6) {
 223                 struct sockaddr_in6 *sin = (struct sockaddr_in6 *)sa;
 224                 printk(KERN_INFO "%s %pi6:%d",
 225                         str, &sin->sin6_addr, ntohs(sin->sin6_port));
 226         }
 227 }
 228
 229 void dst_state_exit_connected(struct dst_state *st)
 230 {
 231         if (st->socket) {
 232                 dst_poll_exit(st);
 233                 st->socket->ops->shutdown(st->socket, 2);
 234
 235                 dst_dump_addr(st->socket, (struct sockaddr *)&st->ctl.addr,
 236                                 "Disconnected peer");
 237                 dst_state_socket_release(st);
 238         }
 239 }
 240
 241 static int dst_state_init_connected(struct dst_state *st)
 242 {
 243         int err;
 244         struct dst_network_ctl *ctl = &st->ctl;
 245
 246         err = dst_state_socket_create(st);
 247         if (err)
 248                 goto err_out_exit;
 249
 250         err = kernel_connect(st->socket, (struct sockaddr *)&st->ctl.addr,
 251                         st->ctl.addr.sa_data_len, 0);
 252         if (err)
 253                 goto err_out_release;
 254
 255         err = dst_poll_init(st);
 256         if (err)
 257                 goto err_out_release;
 258
 259         dst_dump_addr(st->socket, (struct sockaddr *)&ctl->addr,
 260                         "Connected to peer");
 261
 262         return 0;
 263
 264 err_out_release:
 265         dst_state_socket_release(st);
 266 err_out_exit:
 267         return err;
 268 }
 269
 270 /*
 271  * State reset is used to reconnect to the remote peer.
 272  * May fail, but who cares, we will try again later.
 273  */
 274 static inline void dst_state_reset_nolock(struct dst_state *st)
 275 {
 276         dst_state_exit_connected(st);
 277         dst_state_init_connected(st);
 278 }
 279
 280 static inline void dst_state_reset(struct dst_state *st)
 281 {
 282         dst_state_lock(st);
 283         dst_state_reset_nolock(st);
 284         dst_state_unlock(st);
 285 }
 286
 287 /*
 288  * Basic network sending/receiving functions.
 289  * Blocked mode is used.
 290  */
 291 static int dst_data_recv_raw(struct dst_state *st, void *buf, u64 size)
 292 {
 293         struct msghdr msg;
 294         struct kvec iov;
 295         int err;
 296
 297         BUG_ON(!size);
 298
 299         iov.iov_base = buf;
 300         iov.iov_len = size;
 301
 302         msg.msg_iov = (struct iovec *)&iov;
 303         msg.msg_iovlen = 1;
 304         msg.msg_name = NULL;
 305         msg.msg_namelen = 0;
 306         msg.msg_control = NULL;
 307         msg.msg_controllen = 0;
 308         msg.msg_flags = MSG_DONTWAIT;
 309
 310         err = kernel_recvmsg(st->socket, &msg, &iov, 1, iov.iov_len,
 311                         msg.msg_flags);
 312         if (err <= 0) {
 313                 dprintk("%s: failed to recv data: size: %llu, err: %d.\n",
 314                                 __func__, size, err);
 315                 if (err == 0)
 316                         err = -ECONNRESET;
 317
 318                 dst_state_exit_connected(st);
 319         }
 320
 321         return err;
 322 }
 323
 324 /*
 325  * Ping command to early detect failed nodes.
 326  */
 327 static int dst_send_ping(struct dst_state *st)
 328 {
 329         struct dst_cmd *cmd = st->data;
 330         int err = -ECONNRESET;
 331
 332         dst_state_lock(st);
 333         if (st->socket) {
 334                 memset(cmd, 0, sizeof(struct dst_cmd));
 335
 336                 cmd->cmd = __cpu_to_be32(DST_PING);
 337
 338                 err = dst_data_send_header(st->socket, cmd,
 339                                 sizeof(struct dst_cmd), 0);
 340         }
 341         dprintk("%s: st: %p, socket: %p, err: %d.\n", __func__,
 342                         st, st->socket, err);
 343         dst_state_unlock(st);
 344
 345         return err;
 346 }
 347
 348 /*
 349  * Receiving function, which should either return error or read
 350  * whole block request. If there was no traffic for a one second,
 351  * send a ping, since remote node may die.
 352  */
 353 int dst_data_recv(struct dst_state *st, void *data, unsigned int size)
 354 {
 355         unsigned int revents = 0;
 356         unsigned int err_mask = POLLERR | POLLHUP | POLLRDHUP;
 357         unsigned int mask = err_mask | POLLIN;
 358         struct dst_node *n = st->node;
 359         int err = 0;
 360
 361         while (size && !err) {
 362                 revents = dst_state_poll(st);
 363
 364                 if (!(revents & mask)) {
 365                         DEFINE_WAIT(wait);
 366
 367                         for (;;) {
 368                                 prepare_to_wait(&st->thread_wait, &wait,
 369                                                 TASK_INTERRUPTIBLE);
 370                                 if (!n->trans_scan_timeout || st->need_exit)
 371                                         break;
 372
 373                                 revents = dst_state_poll(st);
 374
 375                                 if (revents & mask)
 376                                         break;
 377
 378                                 if (signal_pending(current))
 379                                         break;
 380
 381                                 if (!schedule_timeout(HZ)) {
 382                                         err = dst_send_ping(st);
 383                                         if (err)
 384                                                 return err;
 385                                 }
 386
 387                                 continue;
 388                         }
 389                         finish_wait(&st->thread_wait, &wait);
 390                 }
 391
 392                 err = -ECONNRESET;
 393                 dst_state_lock(st);
 394
 395                 if (st->socket && (st->read_socket == st->socket) &&
 396                                 (revents & POLLIN)) {
 397                         err = dst_data_recv_raw(st, data, size);
 398                         if (err > 0) {
 399                                 data += err;
 400                                 size -= err;
 401                                 err = 0;
 402                         }
 403                 }
 404
 405                 if (revents & err_mask || !st->socket) {
 406                         dprintk("%s: revents: %x, socket: %p, size: %u, "
 407                                         "err: %d.\n", __func__, revents,
 408                                         st->socket, size, err);
 409                         err = -ECONNRESET;
 410                 }
 411
 412                 dst_state_unlock(st);
 413
 414                 if (!n->trans_scan_timeout)
 415                         err = -ENODEV;
 416         }
 417
 418         return err;
 419 }
 420
 421 /*
 422  * Send block autoconf reply.
 423  */
 424 static int dst_process_cfg(struct dst_state *st)
 425 {
 426         struct dst_node *n = st->node;
 427         struct dst_cmd *cmd = st->data;
 428         int err;
 429
 430         cmd->sector = n->size;
 431         cmd->rw = st->permissions;
 432
 433         dst_convert_cmd(cmd);
 434
 435         dst_state_lock(st);
 436         err = dst_data_send_header(st->socket, cmd, sizeof(struct dst_cmd), 0);
 437         dst_state_unlock(st);
 438
 439         return err;
 440 }
 441
 442 /*
 443  * Receive block IO from the network.
 444  */
 445 static int dst_recv_bio(struct dst_state *st, struct bio *bio,
 446                 unsigned int total_size)
 447 {
 448         struct bio_vec *bv;
 449         int i, err;
 450         void *data;
 451         unsigned int sz;
 452
 453         bio_for_each_segment(bv, bio, i) {
 454                 sz = min(total_size, bv->bv_len);
 455
 456                 dprintk("%s: bio: %llu/%u, total: %u, len: %u, sz: %u, "
 457                                 "off: %u.\n", __func__, (u64)bio->bi_sector,
 458                                 bio->bi_size, total_size, bv->bv_len, sz,
 459                                 bv->bv_offset);
 460
 461                 data = kmap(bv->bv_page) + bv->bv_offset;
 462                 err = dst_data_recv(st, data, sz);
 463                 kunmap(bv->bv_page);
 464
 465                 bv->bv_len = sz;
 466
 467                 if (err)
 468                         return err;
 469
 470                 total_size -= sz;
 471                 if (total_size == 0)
 472                         break;
 473         }
 474
 475         return 0;
 476 }
 477
 478 /*
 479  * Our block IO has just completed and arrived: get it.
 480  */
 481 static int dst_process_io_response(struct dst_state *st)
 482 {
 483         struct dst_node *n = st->node;
 484         struct dst_cmd *cmd = st->data;
 485         struct dst_trans *t;
 486         int err = 0;
 487         struct bio *bio;
 488
 489         mutex_lock(&n->trans_lock);
 490         t = dst_trans_search(n, cmd->id);
 491         mutex_unlock(&n->trans_lock);
 492
 493         if (!t)
 494                 goto err_out_exit;
 495
 496         bio = t->bio;
 497
 498         dprintk("%s: bio: %llu/%u, cmd_size: %u, csize: %u, dir: %lu.\n",
 499                 __func__, (u64)bio->bi_sector, bio->bi_size, cmd->size,
 500                 cmd->csize, bio_data_dir(bio));
 501
 502         if (bio_data_dir(bio) == READ) {
 503                 if (bio->bi_size != cmd->size - cmd->csize)
 504                         goto err_out_exit;
 505
 506                 if (dst_need_crypto(n)) {
 507                         err = dst_recv_cdata(st, t->cmd.hash);
 508                         if (err)
 509                                 goto err_out_exit;
 510                 }
 511
 512                 err = dst_recv_bio(st, t->bio, bio->bi_size);
 513                 if (err)
 514                         goto err_out_exit;
 515
 516                 if (dst_need_crypto(n))
 517                         return dst_trans_crypto(t);
 518         } else {
 519                 err = -EBADMSG;
 520                 if (cmd->size || cmd->csize)
 521                         goto err_out_exit;
 522         }
 523
 524         dst_trans_remove(t);
 525         dst_trans_put(t);
 526
 527         return 0;
 528
 529 err_out_exit:
 530         return err;
 531 }
 532
 533 /*
 534  * Receive crypto data.
 535  */
 536 int dst_recv_cdata(struct dst_state *st, void *cdata)
 537 {
 538         struct dst_cmd *cmd = st->data;
 539         struct dst_node *n = st->node;
 540         struct dst_crypto_ctl *c = &n->crypto;
 541         int err;
 542
 543         if (cmd->csize != c->crypto_attached_size) {
 544                 dprintk("%s: cmd: cmd: %u, sector: %llu, size: %u, "
 545                                 "csize: %u != digest size %u.\n",
 546                                 __func__, cmd->cmd, cmd->sector, cmd->size,
 547                                 cmd->csize, c->crypto_attached_size);
 548                 err = -EINVAL;
 549                 goto err_out_exit;
 550         }
 551
 552         err = dst_data_recv(st, cdata, cmd->csize);
 553         if (err)
 554                 goto err_out_exit;
 555
 556         cmd->size -= cmd->csize;
 557         return 0;
 558
 559 err_out_exit:
 560         return err;
 561 }
 562
 563 /*
 564  * Receive the command and start its processing.
 565  */
 566 static int dst_recv_processing(struct dst_state *st)
 567 {
 568         int err = -EINTR;
 569         struct dst_cmd *cmd = st->data;
 570
 571         /*
 572          * If socket will be reset after this statement, then
 573          * dst_data_recv() will just fail and loop will
 574          * start again, so it can be done without any locks.
 575          *
 576          * st->read_socket is needed to prevents state machine
 577          * breaking between this data reading and subsequent one
 578          * in protocol specific functions during connection reset.
 579          * In case of reset we have to read next command and do
 580          * not expect data for old command to magically appear in
 581          * new connection.
 582          */
 583         st->read_socket = st->socket;
 584         err = dst_data_recv(st, cmd, sizeof(struct dst_cmd));
 585         if (err)
 586                 goto out_exit;
 587
 588         dst_convert_cmd(cmd);
 589
 590         dprintk("%s: cmd: %u, size: %u, csize: %u, id: %llu, "
 591                         "sector: %llu, flags: %llx, rw: %llx.\n",
 592                         __func__, cmd->cmd, cmd->size,
 593                         cmd->csize, cmd->id, cmd->sector,
 594                         cmd->flags, cmd->rw);
 595
 596         /*
 597          * This should catch protocol breakage and random garbage
 598          * instead of commands.
 599          */
 600         if (unlikely(cmd->csize > st->size - sizeof(struct dst_cmd))) {
 601                 err = -EBADMSG;
 602                 goto out_exit;
 603         }
 604
 605         err = -EPROTO;
 606         switch (cmd->cmd) {
 607         case DST_IO_RESPONSE:
 608                 err = dst_process_io_response(st);
 609                 break;
 610         case DST_IO:
 611                 err = dst_process_io(st);
 612                 break;
 613         case DST_CFG:
 614                 err = dst_process_cfg(st);
 615                 break;
 616         case DST_PING:
 617                 err = 0;
 618                 break;
 619         default:
 620                 break;
 621         }
 622
 623 out_exit:
 624         return err;
 625 }
 626
 627 /*
 628  * Receiving thread. For the client node we should try to reconnect,
 629  * for accepted client we just drop the state and expect it to reconnect.
 630  */
 631 static int dst_recv(void *init_data, void *schedule_data)
 632 {
 633         struct dst_state *st = schedule_data;
 634         struct dst_node *n = init_data;
 635         int err = 0;
 636
 637         dprintk("%s: start st: %p, n: %p, scan: %lu, need_exit: %d.\n",
 638                         __func__, st, n, n->trans_scan_timeout, st->need_exit);
 639
 640         while (n->trans_scan_timeout && !st->need_exit) {
 641                 err = dst_recv_processing(st);
 642                 if (err < 0) {
 643                         if (!st->ctl.type)
 644                                 break;
 645
 646                         if (!n->trans_scan_timeout || st->need_exit)
 647                                 break;
 648
 649                         dst_state_reset(st);
 650                         msleep(1000);
 651                 }
 652         }
 653
 654         st->need_exit = 1;
 655         wake_up(&st->thread_wait);
 656
 657         dprintk("%s: freeing receiving socket st: %p.\n", __func__, st);
 658         dst_state_lock(st);
 659         dst_state_exit_connected(st);
 660         dst_state_unlock(st);
 661         dst_state_put(st);
 662
 663         dprintk("%s: freed receiving socket st: %p.\n", __func__, st);
 664
 665         return err;
 666 }
 667
 668 /*
 669  * Network state dies here and borns couple of lines below.
 670  * This object is the main network state processing engine:
 671  * sending, receiving, reconnections, all network related
 672  * tasks are handled on behalf of the state.
 673  */
 674 static void dst_state_free(struct dst_state *st)
 675 {
 676         dprintk("%s: st: %p.\n", __func__, st);
 677         if (st->cleanup)
 678                 st->cleanup(st);
 679         kfree(st->data);
 680         kfree(st);
 681 }
 682
 683 struct dst_state *dst_state_alloc(struct dst_node *n)
 684 {
 685         struct dst_state *st;
 686         int err = -ENOMEM;
 687
 688         st = kzalloc(sizeof(struct dst_state), GFP_KERNEL);
 689         if (!st)
 690                 goto err_out_exit;
 691
 692         st->node = n;
 693         st->need_exit = 0;
 694
 695         st->size = PAGE_SIZE;
 696         st->data = kmalloc(st->size, GFP_KERNEL);
 697         if (!st->data)
 698                 goto err_out_free;
 699
 700         spin_lock_init(&st->request_lock);
 701         INIT_LIST_HEAD(&st->request_list);
 702
 703         mutex_init(&st->state_lock);
 704         init_waitqueue_head(&st->thread_wait);
 705
 706         /*
 707          * One for processing thread, another one for node itself.
 708          */
 709         atomic_set(&st->refcnt, 2);
 710
 711         dprintk("%s: st: %p, n: %p.\n", __func__, st, st->node);
 712
 713         return st;
 714
 715 err_out_free:
 716         kfree(st);
 717 err_out_exit:
 718         return ERR_PTR(err);
 719 }
 720
 721 int dst_state_schedule_receiver(struct dst_state *st)
 722 {
 723         return thread_pool_schedule_private(st->node->pool, dst_thread_setup,
 724                         dst_recv, st, MAX_SCHEDULE_TIMEOUT, st->node);
 725 }
 726
 727 /*
 728  * Initialize client's connection to the remote peer: allocate state,
 729  * connect and perform block IO autoconfiguration.
 730  */
 731 int dst_node_init_connected(struct dst_node *n, struct dst_network_ctl *r)
 732 {
 733         struct dst_state *st;
 734         int err = -ENOMEM;
 735
 736         st = dst_state_alloc(n);
 737         if (IS_ERR(st)) {
 738                 err = PTR_ERR(st);
 739                 goto err_out_exit;
 740         }
 741         memcpy(&st->ctl, r, sizeof(struct dst_network_ctl));
 742
 743         err = dst_state_init_connected(st);
 744         if (err)
 745                 goto err_out_free_data;
 746
 747         err = dst_request_remote_config(st);
 748         if (err)
 749                 goto err_out_exit_connected;
 750         n->state = st;
 751
 752         err = dst_state_schedule_receiver(st);
 753         if (err)
 754                 goto err_out_exit_connected;
 755
 756         return 0;
 757
 758 err_out_exit_connected:
 759         dst_state_exit_connected(st);
 760 err_out_free_data:
 761         dst_state_free(st);
 762 err_out_exit:
 763         n->state = NULL;
 764         return err;
 765 }
 766
 767 void dst_state_put(struct dst_state *st)
 768 {
 769         dprintk("%s: st: %p, refcnt: %d.\n",
 770                         __func__, st, atomic_read(&st->refcnt));
 771         if (atomic_dec_and_test(&st->refcnt))
 772                 dst_state_free(st);
 773 }
 774
 775 /*
 776  * Send block IO to the network one by one using zero-copy ->sendpage().
 777  */
 778 int dst_send_bio(struct dst_state *st, struct dst_cmd *cmd, struct bio *bio)
 779 {
 780         struct bio_vec *bv;
 781         struct dst_crypto_ctl *c = &st->node->crypto;
 782         int err, i = 0;
 783         int flags = MSG_WAITALL;
 784
 785         err = dst_data_send_header(st->socket, cmd,
 786                 sizeof(struct dst_cmd) + c->crypto_attached_size, bio->bi_vcnt);
 787         if (err)
 788                 goto err_out_exit;
 789
 790         bio_for_each_segment(bv, bio, i) {
 791                 if (i < bio->bi_vcnt - 1)
 792                         flags |= MSG_MORE;
 793
 794                 err = kernel_sendpage(st->socket, bv->bv_page, bv->bv_offset,
 795                                 bv->bv_len, flags);
 796                 if (err <= 0)
 797                         goto err_out_exit;
 798         }
 799
 800         return 0;
 801
 802 err_out_exit:
 803         dprintk("%s: %d/%d, flags: %x, err: %d.\n",
 804                         __func__, i, bio->bi_vcnt, flags, err);
 805         return err;
 806 }
 807
 808 /*
 809  * Send transaction to the remote peer.
 810  */
 811 int dst_trans_send(struct dst_trans *t)
 812 {
 813         int err;
 814         struct dst_state *st = t->n->state;
 815         struct bio *bio = t->bio;
 816
 817         dst_convert_cmd(&t->cmd);
 818
 819         dst_state_lock(st);
 820         if (!st->socket) {
 821                 err = dst_state_init_connected(st);
 822                 if (err)
 823                         goto err_out_unlock;
 824         }
 825
 826         if (bio_data_dir(bio) == WRITE) {
 827                 err = dst_send_bio(st, &t->cmd, t->bio);
 828         } else {
 829                 err = dst_data_send_header(st->socket, &t->cmd,
 830                                 sizeof(struct dst_cmd), 0);
 831         }
 832         if (err)
 833                 goto err_out_reset;
 834
 835         dst_state_unlock(st);
 836         return 0;
 837
 838 err_out_reset:
 839         dst_state_reset_nolock(st);
 840 err_out_unlock:
 841         dst_state_unlock(st);
 842
 843         return err;
 844 }