2 * 2007+ Copyright (c) Evgeniy Polyakov <zbr@ioremap.net>
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.
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.
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>
23 #include <linux/in6.h>
24 #include <linux/socket.h>
25 #include <linux/slab.h>
33 struct dst_poll_helper
39 static int dst_queue_wake(wait_queue_t *wait, unsigned mode, int sync, void *key)
41 struct dst_state *st = container_of(wait, struct dst_state, wait);
43 wake_up(&st->thread_wait);
47 static void dst_queue_func(struct file *file, wait_queue_head_t *whead,
50 struct dst_state *st = container_of(pt, struct dst_poll_helper, pt)->st;
53 init_waitqueue_func_entry(&st->wait, dst_queue_wake);
54 add_wait_queue(whead, &st->wait);
57 void dst_poll_exit(struct dst_state *st)
60 remove_wait_queue(st->whead, &st->wait);
65 int dst_poll_init(struct dst_state *st)
67 struct dst_poll_helper ph;
70 init_poll_funcptr(&ph.pt, &dst_queue_func);
72 st->socket->ops->poll(NULL, st->socket, &ph.pt);
77 * Header receiving function - may block.
79 static int dst_data_recv_header(struct socket *sock,
80 void *data, unsigned int size, int block)
89 msg.msg_iov = (struct iovec *)&iov;
93 msg.msg_control = NULL;
94 msg.msg_controllen = 0;
95 msg.msg_flags = (block)?MSG_WAITALL:MSG_DONTWAIT;
97 err = kernel_recvmsg(sock, &msg, &iov, 1, iov.iov_len,
106 * Header sending function - may block.
108 int dst_data_send_header(struct socket *sock,
109 void *data, unsigned int size, int more)
118 msg.msg_iov = (struct iovec *)&iov;
122 msg.msg_control = NULL;
123 msg.msg_controllen = 0;
124 msg.msg_flags = MSG_WAITALL | (more)?MSG_MORE:0;
126 err = kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
128 dprintk("%s: size: %u, more: %d, err: %d.\n",
129 __func__, size, more, err);
137 * Block autoconfiguration: request size of the storage and permissions.
139 static int dst_request_remote_config(struct dst_state *st)
141 struct dst_node *n = st->node;
143 struct dst_cmd *cmd = st->data;
145 memset(cmd, 0, sizeof(struct dst_cmd));
148 dst_convert_cmd(cmd);
150 err = dst_data_send_header(st->socket, cmd, sizeof(struct dst_cmd), 0);
154 err = dst_data_recv_header(st->socket, cmd, sizeof(struct dst_cmd), 1);
158 dst_convert_cmd(cmd);
160 if (cmd->cmd != DST_CFG) {
162 dprintk("%s: checking result: cmd: %d, size reported: %llu.\n",
163 __func__, cmd->cmd, cmd->sector);
168 n->size = min_t(loff_t, n->size, cmd->sector);
170 n->size = cmd->sector;
172 n->info->size = n->size;
173 st->permissions = cmd->rw;
176 dprintk("%s: n: %p, err: %d, size: %llu, permission: %x.\n",
177 __func__, n, err, n->size, st->permissions);
185 #define DST_DEFAULT_TIMEO 20000
187 int dst_state_socket_create(struct dst_state *st)
191 struct dst_network_ctl *ctl = &st->ctl;
193 err = sock_create(ctl->addr.sa_family, ctl->type, ctl->proto, &sock);
197 sock->sk->sk_sndtimeo = sock->sk->sk_rcvtimeo =
198 msecs_to_jiffies(DST_DEFAULT_TIMEO);
199 sock->sk->sk_allocation = GFP_NOIO;
201 st->socket = st->read_socket = sock;
205 void dst_state_socket_release(struct dst_state *st)
207 dprintk("%s: st: %p, socket: %p, n: %p.\n",
208 __func__, st, st->socket, st->node);
210 sock_release(st->socket);
212 st->read_socket = NULL;
216 void dst_dump_addr(struct socket *sk, struct sockaddr *sa, char *str)
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",
221 str, 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));
229 void dst_state_exit_connected(struct dst_state *st)
233 st->socket->ops->shutdown(st->socket, 2);
235 dst_dump_addr(st->socket, (struct sockaddr *)&st->ctl.addr,
236 "Disconnected peer");
237 dst_state_socket_release(st);
241 static int dst_state_init_connected(struct dst_state *st)
244 struct dst_network_ctl *ctl = &st->ctl;
246 err = dst_state_socket_create(st);
250 err = kernel_connect(st->socket, (struct sockaddr *)&st->ctl.addr,
251 st->ctl.addr.sa_data_len, 0);
253 goto err_out_release;
255 err = dst_poll_init(st);
257 goto err_out_release;
259 dst_dump_addr(st->socket, (struct sockaddr *)&ctl->addr,
260 "Connected to peer");
265 dst_state_socket_release(st);
271 * State reset is used to reconnect to the remote peer.
272 * May fail, but who cares, we will try again later.
274 static void inline dst_state_reset_nolock(struct dst_state *st)
276 dst_state_exit_connected(st);
277 dst_state_init_connected(st);
280 static void inline dst_state_reset(struct dst_state *st)
283 dst_state_reset_nolock(st);
284 dst_state_unlock(st);
288 * Basic network sending/receiving functions.
289 * Blocked mode is used.
291 static int dst_data_recv_raw(struct dst_state *st, void *buf, u64 size)
302 msg.msg_iov = (struct iovec *)&iov;
306 msg.msg_control = NULL;
307 msg.msg_controllen = 0;
308 msg.msg_flags = MSG_DONTWAIT;
310 err = kernel_recvmsg(st->socket, &msg, &iov, 1, iov.iov_len,
313 dprintk("%s: failed to recv data: size: %llu, err: %d.\n",
314 __func__, size, err);
318 dst_state_exit_connected(st);
325 * Ping command to early detect failed nodes.
327 static int dst_send_ping(struct dst_state *st)
329 struct dst_cmd *cmd = st->data;
330 int err = -ECONNRESET;
334 memset(cmd, 0, sizeof(struct dst_cmd));
336 cmd->cmd = __cpu_to_be32(DST_PING);
338 err = dst_data_send_header(st->socket, cmd, sizeof(struct dst_cmd), 0);
340 dprintk("%s: st: %p, socket: %p, err: %d.\n", __func__, st, st->socket, err);
341 dst_state_unlock(st);
347 * Receiving function, which should either return error or read
348 * whole block request. If there was no traffic for a one second,
349 * send a ping, since remote node may die.
351 int dst_data_recv(struct dst_state *st, void *data, unsigned int size)
353 unsigned int revents = 0;
354 unsigned int err_mask = POLLERR | POLLHUP | POLLRDHUP;
355 unsigned int mask = err_mask | POLLIN;
356 struct dst_node *n = st->node;
359 while (size && !err) {
360 revents = dst_state_poll(st);
362 if (!(revents & mask)) {
366 prepare_to_wait(&st->thread_wait, &wait,
368 if (!n->trans_scan_timeout || st->need_exit)
371 revents = dst_state_poll(st);
376 if (signal_pending(current))
379 if (!schedule_timeout(HZ)) {
380 err = dst_send_ping(st);
387 finish_wait(&st->thread_wait, &wait);
394 (st->read_socket == st->socket) &&
395 (revents & POLLIN)) {
396 err = dst_data_recv_raw(st, data, size);
404 if (revents & err_mask || !st->socket) {
405 dprintk("%s: revents: %x, socket: %p, size: %u, err: %d.\n",
406 __func__, revents, st->socket, size, err);
410 dst_state_unlock(st);
412 if (!n->trans_scan_timeout)
420 * Send block autoconf reply.
422 static int dst_process_cfg(struct dst_state *st)
424 struct dst_node *n = st->node;
425 struct dst_cmd *cmd = st->data;
428 cmd->sector = n->size;
429 cmd->rw = st->permissions;
431 dst_convert_cmd(cmd);
434 err = dst_data_send_header(st->socket, cmd, sizeof(struct dst_cmd), 0);
435 dst_state_unlock(st);
441 * Receive block IO from the network.
443 static int dst_recv_bio(struct dst_state *st, struct bio *bio, unsigned int total_size)
450 bio_for_each_segment(bv, bio, i) {
451 sz = min(total_size, bv->bv_len);
453 dprintk("%s: bio: %llu/%u, total: %u, len: %u, sz: %u, off: %u.\n",
454 __func__, (u64)bio->bi_sector, bio->bi_size, total_size,
455 bv->bv_len, sz, bv->bv_offset);
457 data = kmap(bv->bv_page) + bv->bv_offset;
458 err = dst_data_recv(st, data, sz);
475 * Our block IO has just completed and arrived: get it.
477 static int dst_process_io_response(struct dst_state *st)
479 struct dst_node *n = st->node;
480 struct dst_cmd *cmd = st->data;
485 mutex_lock(&n->trans_lock);
486 t = dst_trans_search(n, cmd->id);
487 mutex_unlock(&n->trans_lock);
494 dprintk("%s: bio: %llu/%u, cmd_size: %u, csize: %u, dir: %lu.\n",
495 __func__, (u64)bio->bi_sector, bio->bi_size, cmd->size,
496 cmd->csize, bio_data_dir(bio));
498 if (bio_data_dir(bio) == READ) {
499 if (bio->bi_size != cmd->size - cmd->csize)
502 if (dst_need_crypto(n)) {
503 err = dst_recv_cdata(st, t->cmd.hash);
508 err = dst_recv_bio(st, t->bio, bio->bi_size);
512 if (dst_need_crypto(n))
513 return dst_trans_crypto(t);
516 if (cmd->size || cmd->csize)
530 * Receive crypto data.
532 int dst_recv_cdata(struct dst_state *st, void *cdata)
534 struct dst_cmd *cmd = st->data;
535 struct dst_node *n = st->node;
536 struct dst_crypto_ctl *c = &n->crypto;
539 if (cmd->csize != c->crypto_attached_size) {
540 dprintk("%s: cmd: cmd: %u, sector: %llu, size: %u, "
541 "csize: %u != digest size %u.\n",
542 __func__, cmd->cmd, cmd->sector, cmd->size,
543 cmd->csize, c->crypto_attached_size);
548 err = dst_data_recv(st, cdata, cmd->csize);
552 cmd->size -= cmd->csize;
560 * Receive the command and start its processing.
562 static int dst_recv_processing(struct dst_state *st)
565 struct dst_cmd *cmd = st->data;
568 * If socket will be reset after this statement, then
569 * dst_data_recv() will just fail and loop will
570 * start again, so it can be done without any locks.
572 * st->read_socket is needed to prevents state machine
573 * breaking between this data reading and subsequent one
574 * in protocol specific functions during connection reset.
575 * In case of reset we have to read next command and do
576 * not expect data for old command to magically appear in
579 st->read_socket = st->socket;
580 err = dst_data_recv(st, cmd, sizeof(struct dst_cmd));
584 dst_convert_cmd(cmd);
586 dprintk("%s: cmd: %u, size: %u, csize: %u, id: %llu, "
587 "sector: %llu, flags: %llx, rw: %llx.\n",
588 __func__, cmd->cmd, cmd->size,
589 cmd->csize, cmd->id, cmd->sector,
590 cmd->flags, cmd->rw);
593 * This should catch protocol breakage and random garbage instead of commands.
595 if (unlikely(cmd->csize > st->size - sizeof(struct dst_cmd))) {
602 case DST_IO_RESPONSE:
603 err = dst_process_io_response(st);
606 err = dst_process_io(st);
609 err = dst_process_cfg(st);
623 * Receiving thread. For the client node we should try to reconnect,
624 * for accepted client we just drop the state and expect it to reconnect.
626 static int dst_recv(void *init_data, void *schedule_data)
628 struct dst_state *st = schedule_data;
629 struct dst_node *n = init_data;
632 dprintk("%s: start st: %p, n: %p, scan: %lu, need_exit: %d.\n",
633 __func__, st, n, n->trans_scan_timeout, st->need_exit);
635 while (n->trans_scan_timeout && !st->need_exit) {
636 err = dst_recv_processing(st);
641 if (!n->trans_scan_timeout || st->need_exit)
650 wake_up(&st->thread_wait);
652 dprintk("%s: freeing receiving socket st: %p.\n", __func__, st);
654 dst_state_exit_connected(st);
655 dst_state_unlock(st);
658 dprintk("%s: freed receiving socket st: %p.\n", __func__, st);
664 * Network state dies here and borns couple of lines below.
665 * This object is the main network state processing engine:
666 * sending, receiving, reconnections, all network related
667 * tasks are handled on behalf of the state.
669 static void dst_state_free(struct dst_state *st)
671 dprintk("%s: st: %p.\n", __func__, st);
678 struct dst_state *dst_state_alloc(struct dst_node *n)
680 struct dst_state *st;
683 st = kzalloc(sizeof(struct dst_state), GFP_KERNEL);
690 st->size = PAGE_SIZE;
691 st->data = kmalloc(st->size, GFP_KERNEL);
695 spin_lock_init(&st->request_lock);
696 INIT_LIST_HEAD(&st->request_list);
698 mutex_init(&st->state_lock);
699 init_waitqueue_head(&st->thread_wait);
702 * One for processing thread, another one for node itself.
704 atomic_set(&st->refcnt, 2);
706 dprintk("%s: st: %p, n: %p.\n", __func__, st, st->node);
716 int dst_state_schedule_receiver(struct dst_state *st)
718 return thread_pool_schedule_private(st->node->pool, dst_thread_setup,
719 dst_recv, st, MAX_SCHEDULE_TIMEOUT, st->node);
723 * Initialize client's connection to the remote peer: allocate state,
724 * connect and perform block IO autoconfiguration.
726 int dst_node_init_connected(struct dst_node *n, struct dst_network_ctl *r)
728 struct dst_state *st;
731 st = dst_state_alloc(n);
736 memcpy(&st->ctl, r, sizeof(struct dst_network_ctl));
738 err = dst_state_init_connected(st);
740 goto err_out_free_data;
742 err = dst_request_remote_config(st);
744 goto err_out_exit_connected;
747 err = dst_state_schedule_receiver(st);
749 goto err_out_exit_connected;
753 err_out_exit_connected:
754 dst_state_exit_connected(st);
762 void dst_state_put(struct dst_state *st)
764 dprintk("%s: st: %p, refcnt: %d.\n",
765 __func__, st, atomic_read(&st->refcnt));
766 if (atomic_dec_and_test(&st->refcnt))
771 * Send block IO to the network one by one using zero-copy ->sendpage().
773 int dst_send_bio(struct dst_state *st, struct dst_cmd *cmd, struct bio *bio)
776 struct dst_crypto_ctl *c = &st->node->crypto;
778 int flags = MSG_WAITALL;
780 err = dst_data_send_header(st->socket, cmd,
781 sizeof(struct dst_cmd) + c->crypto_attached_size, bio->bi_vcnt);
785 bio_for_each_segment(bv, bio, i) {
786 if (i < bio->bi_vcnt - 1)
789 err = kernel_sendpage(st->socket, bv->bv_page, bv->bv_offset,
798 dprintk("%s: %d/%d, flags: %x, err: %d.\n",
799 __func__, i, bio->bi_vcnt, flags, err);
804 * Send transaction to the remote peer.
806 int dst_trans_send(struct dst_trans *t)
809 struct dst_state *st = t->n->state;
810 struct bio *bio = t->bio;
812 dst_convert_cmd(&t->cmd);
816 err = dst_state_init_connected(st);
821 if (bio_data_dir(bio) == WRITE) {
822 err = dst_send_bio(st, &t->cmd, t->bio);
824 err = dst_data_send_header(st->socket, &t->cmd,
825 sizeof(struct dst_cmd), 0);
830 dst_state_unlock(st);
834 dst_state_reset_nolock(st);
836 dst_state_unlock(st);