1/* -*- mode: c; c-basic-offset: 8; -*- 2 * 3 * vim: noexpandtab sw=8 ts=8 sts=0: 4 * 5 * Copyright (C) 2004 Oracle. All rights reserved. 6 * 7 * This program is free software; you can redistribute it and/or 8 * modify it under the terms of the GNU General Public 9 * License as published by the Free Software Foundation; either 10 * version 2 of the License, or (at your option) any later version. 11 * 12 * This program is distributed in the hope that it will be useful, 13 * but WITHOUT ANY WARRANTY; without even the implied warranty of 14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 15 * General Public License for more details. 16 * 17 * You should have received a copy of the GNU General Public 18 * License along with this program; if not, write to the 19 * Free Software Foundation, Inc., 59 Temple Place - Suite 330, 20 * Boston, MA 021110-1307, USA. 21 * 22 * ---- 23 * 24 * Callers for this were originally written against a very simple synchronus 25 * API. This implementation reflects those simple callers. Some day I'm sure 26 * we'll need to move to a more robust posting/callback mechanism. 27 * 28 * Transmit calls pass in kernel virtual addresses and block copying this into 29 * the socket's tx buffers via a usual blocking sendmsg. They'll block waiting 30 * for a failed socket to timeout. TX callers can also pass in a poniter to an 31 * 'int' which gets filled with an errno off the wire in response to the 32 * message they send. 33 * 34 * Handlers for unsolicited messages are registered. Each socket has a page 35 * that incoming data is copied into. First the header, then the data. 36 * Handlers are called from only one thread with a reference to this per-socket 37 * page. This page is destroyed after the handler call, so it can't be 38 * referenced beyond the call. Handlers may block but are discouraged from 39 * doing so. 40 * 41 * Any framing errors (bad magic, large payload lengths) close a connection. 42 * 43 * Our sock_container holds the state we associate with a socket. It's current 44 * framing state is held there as well as the refcounting we do around when it 45 * is safe to tear down the socket. The socket is only finally torn down from 46 * the container when the container loses all of its references -- so as long 47 * as you hold a ref on the container you can trust that the socket is valid 48 * for use with kernel socket APIs. 49 * 50 * Connections are initiated between a pair of nodes when the node with the 51 * higher node number gets a heartbeat callback which indicates that the lower 52 * numbered node has started heartbeating. The lower numbered node is passive 53 * and only accepts the connection if the higher numbered node is heartbeating. 54 */ 55 56#include <linux/kernel.h> 57#include <linux/jiffies.h> 58#include <linux/slab.h> 59#include <linux/idr.h> 60#include <linux/kref.h> 61#include <net/tcp.h> 62 63#include <asm/uaccess.h> 64 65#include "heartbeat.h" 66#include "tcp.h" 67#include "nodemanager.h" 68#define MLOG_MASK_PREFIX ML_TCP 69#include "masklog.h" 70#include "quorum.h" 71 72#include "tcp_internal.h" 73 74/* 75 * The linux network stack isn't sparse endian clean.. It has macros like 76 * ntohs() which perform the endian checks and structs like sockaddr_in 77 * which aren't annotated. So __force is found here to get the build 78 * clean. When they emerge from the dark ages and annotate the code 79 * we can remove these. 80 */ 81 82#define SC_NODEF_FMT "node %s (num %u) at %u.%u.%u.%u:%u" 83#define SC_NODEF_ARGS(sc) sc->sc_node->nd_name, sc->sc_node->nd_num, \ 84 NIPQUAD(sc->sc_node->nd_ipv4_address), \ 85 ntohs(sc->sc_node->nd_ipv4_port) 86 87/* 88 * In the following two log macros, the whitespace after the ',' just 89 * before ##args is intentional. Otherwise, gcc 2.95 will eat the 90 * previous token if args expands to nothing. 91 */ 92#define msglog(hdr, fmt, args...) do { \ 93 typeof(hdr) __hdr = (hdr); \ 94 mlog(ML_MSG, "[mag %u len %u typ %u stat %d sys_stat %d " \ 95 "key %08x num %u] " fmt, \ 96 be16_to_cpu(__hdr->magic), be16_to_cpu(__hdr->data_len), \ 97 be16_to_cpu(__hdr->msg_type), be32_to_cpu(__hdr->status), \ 98 be32_to_cpu(__hdr->sys_status), be32_to_cpu(__hdr->key), \ 99 be32_to_cpu(__hdr->msg_num) , ##args); \ 100} while (0) 101 102#define sclog(sc, fmt, args...) do { \ 103 typeof(sc) __sc = (sc); \ 104 mlog(ML_SOCKET, "[sc %p refs %d sock %p node %u page %p " \ 105 "pg_off %zu] " fmt, __sc, \ 106 atomic_read(&__sc->sc_kref.refcount), __sc->sc_sock, \ 107 __sc->sc_node->nd_num, __sc->sc_page, __sc->sc_page_off , \ 108 ##args); \ 109} while (0) 110 111static DEFINE_RWLOCK(o2net_handler_lock); 112static struct rb_root o2net_handler_tree = RB_ROOT; 113 114static struct o2net_node o2net_nodes[O2NM_MAX_NODES]; 115 116static struct socket *o2net_listen_sock = NULL; 117 118/* 119 * listen work is only queued by the listening socket callbacks on the 120 * o2net_wq. teardown detaches the callbacks before destroying the workqueue. 121 * quorum work is queued as sock containers are shutdown.. stop_listening 122 * tears down all the node's sock containers, preventing future shutdowns 123 * and queued quroum work, before canceling delayed quorum work and 124 * destroying the work queue. 125 */ 126static struct workqueue_struct *o2net_wq; 127static struct work_struct o2net_listen_work; 128 129static struct o2hb_callback_func o2net_hb_up, o2net_hb_down; 130#define O2NET_HB_PRI 0x1 131 132static struct o2net_handshake *o2net_hand; 133static struct o2net_msg *o2net_keep_req, *o2net_keep_resp; 134 135static int o2net_sys_err_translations[O2NET_ERR_MAX] = 136 {[O2NET_ERR_NONE] = 0, 137 [O2NET_ERR_NO_HNDLR] = -ENOPROTOOPT, 138 [O2NET_ERR_OVERFLOW] = -EOVERFLOW, 139 [O2NET_ERR_DIED] = -EHOSTDOWN,}; 140 141/* can't quite avoid *all* internal declarations :/ */ 142static void o2net_sc_connect_completed(struct work_struct *work); 143static void o2net_rx_until_empty(struct work_struct *work); 144static void o2net_shutdown_sc(struct work_struct *work); 145static void o2net_listen_data_ready(struct sock *sk, int bytes); 146static void o2net_sc_send_keep_req(struct work_struct *work); 147static void o2net_idle_timer(unsigned long data); 148static void o2net_sc_postpone_idle(struct o2net_sock_container *sc); 149static void o2net_sc_reset_idle_timer(struct o2net_sock_container *sc); 150 151static inline int o2net_reconnect_delay(struct o2nm_node *node) 152{ 153 return o2nm_single_cluster->cl_reconnect_delay_ms; 154} 155 156static inline int o2net_keepalive_delay(struct o2nm_node *node) 157{ 158 return o2nm_single_cluster->cl_keepalive_delay_ms; 159} 160 161static inline int o2net_idle_timeout(struct o2nm_node *node) 162{ 163 return o2nm_single_cluster->cl_idle_timeout_ms; 164} 165 166static inline int o2net_sys_err_to_errno(enum o2net_system_error err) 167{ 168 int trans; 169 BUG_ON(err >= O2NET_ERR_MAX); 170 trans = o2net_sys_err_translations[err]; 171 172 /* Just in case we mess up the translation table above */ 173 BUG_ON(err != O2NET_ERR_NONE && trans == 0); 174 return trans; 175} 176 177static struct o2net_node * o2net_nn_from_num(u8 node_num) 178{ 179 BUG_ON(node_num >= ARRAY_SIZE(o2net_nodes)); 180 return &o2net_nodes[node_num]; 181} 182 183static u8 o2net_num_from_nn(struct o2net_node *nn) 184{ 185 BUG_ON(nn == NULL); 186 return nn - o2net_nodes; 187} 188 189/* ------------------------------------------------------------ */ 190 191static int o2net_prep_nsw(struct o2net_node *nn, struct o2net_status_wait *nsw) 192{ 193 int ret = 0; 194 195 do { 196 if (!idr_pre_get(&nn->nn_status_idr, GFP_ATOMIC)) { 197 ret = -EAGAIN; 198 break; 199 } 200 spin_lock(&nn->nn_lock); 201 ret = idr_get_new(&nn->nn_status_idr, nsw, &nsw->ns_id); 202 if (ret == 0) 203 list_add_tail(&nsw->ns_node_item, 204 &nn->nn_status_list); 205 spin_unlock(&nn->nn_lock); 206 } while (ret == -EAGAIN); 207 208 if (ret == 0) { 209 init_waitqueue_head(&nsw->ns_wq); 210 nsw->ns_sys_status = O2NET_ERR_NONE; 211 nsw->ns_status = 0; 212 } 213 214 return ret; 215} 216 217static void o2net_complete_nsw_locked(struct o2net_node *nn, 218 struct o2net_status_wait *nsw, 219 enum o2net_system_error sys_status, 220 s32 status) 221{ 222 assert_spin_locked(&nn->nn_lock); 223 224 if (!list_empty(&nsw->ns_node_item)) { 225 list_del_init(&nsw->ns_node_item); 226 nsw->ns_sys_status = sys_status; 227 nsw->ns_status = status; 228 idr_remove(&nn->nn_status_idr, nsw->ns_id); 229 wake_up(&nsw->ns_wq); 230 } 231} 232 233static void o2net_complete_nsw(struct o2net_node *nn, 234 struct o2net_status_wait *nsw, 235 u64 id, enum o2net_system_error sys_status, 236 s32 status) 237{ 238 spin_lock(&nn->nn_lock); 239 if (nsw == NULL) { 240 if (id > INT_MAX) 241 goto out; 242 243 nsw = idr_find(&nn->nn_status_idr, id); 244 if (nsw == NULL) 245 goto out; 246 } 247 248 o2net_complete_nsw_locked(nn, nsw, sys_status, status); 249 250out: 251 spin_unlock(&nn->nn_lock); 252 return; 253} 254 255static void o2net_complete_nodes_nsw(struct o2net_node *nn) 256{ 257 struct list_head *iter, *tmp; 258 unsigned int num_kills = 0; 259 struct o2net_status_wait *nsw; 260 261 assert_spin_locked(&nn->nn_lock); 262 263 list_for_each_safe(iter, tmp, &nn->nn_status_list) { 264 nsw = list_entry(iter, struct o2net_status_wait, ns_node_item); 265 o2net_complete_nsw_locked(nn, nsw, O2NET_ERR_DIED, 0); 266 num_kills++; 267 } 268 269 mlog(0, "completed %d messages for node %u\n", num_kills, 270 o2net_num_from_nn(nn)); 271} 272 273static int o2net_nsw_completed(struct o2net_node *nn, 274 struct o2net_status_wait *nsw) 275{ 276 int completed; 277 spin_lock(&nn->nn_lock); 278 completed = list_empty(&nsw->ns_node_item); 279 spin_unlock(&nn->nn_lock); 280 return completed; 281} 282 283/* ------------------------------------------------------------ */ 284 285static void sc_kref_release(struct kref *kref) 286{ 287 struct o2net_sock_container *sc = container_of(kref, 288 struct o2net_sock_container, sc_kref); 289 BUG_ON(timer_pending(&sc->sc_idle_timeout)); 290 291 sclog(sc, "releasing\n"); 292 293 if (sc->sc_sock) { 294 sock_release(sc->sc_sock); 295 sc->sc_sock = NULL; 296 } 297 298 o2nm_node_put(sc->sc_node); 299 sc->sc_node = NULL; 300 301 kfree(sc); 302} 303 304static void sc_put(struct o2net_sock_container *sc) 305{ 306 sclog(sc, "put\n"); 307 kref_put(&sc->sc_kref, sc_kref_release); 308} 309static void sc_get(struct o2net_sock_container *sc) 310{ 311 sclog(sc, "get\n"); 312 kref_get(&sc->sc_kref); 313} 314static struct o2net_sock_container *sc_alloc(struct o2nm_node *node) 315{ 316 struct o2net_sock_container *sc, *ret = NULL; 317 struct page *page = NULL; 318 319 page = alloc_page(GFP_NOFS); 320 sc = kzalloc(sizeof(*sc), GFP_NOFS); 321 if (sc == NULL || page == NULL) 322 goto out; 323 324 kref_init(&sc->sc_kref); 325 o2nm_node_get(node); 326 sc->sc_node = node; 327 328 INIT_WORK(&sc->sc_connect_work, o2net_sc_connect_completed); 329 INIT_WORK(&sc->sc_rx_work, o2net_rx_until_empty); 330 INIT_WORK(&sc->sc_shutdown_work, o2net_shutdown_sc); 331 INIT_DELAYED_WORK(&sc->sc_keepalive_work, o2net_sc_send_keep_req); 332 333 init_timer(&sc->sc_idle_timeout); 334 sc->sc_idle_timeout.function = o2net_idle_timer; 335 sc->sc_idle_timeout.data = (unsigned long)sc; 336 337 sclog(sc, "alloced\n"); 338 339 ret = sc; 340 sc->sc_page = page; 341 sc = NULL; 342 page = NULL; 343 344out: 345 if (page) 346 __free_page(page); 347 kfree(sc); 348 349 return ret; 350} 351 352/* ------------------------------------------------------------ */ 353 354static void o2net_sc_queue_work(struct o2net_sock_container *sc, 355 struct work_struct *work) 356{ 357 sc_get(sc); 358 if (!queue_work(o2net_wq, work)) 359 sc_put(sc); 360} 361static void o2net_sc_queue_delayed_work(struct o2net_sock_container *sc, 362 struct delayed_work *work, 363 int delay) 364{ 365 sc_get(sc); 366 if (!queue_delayed_work(o2net_wq, work, delay)) 367 sc_put(sc); 368} 369static void o2net_sc_cancel_delayed_work(struct o2net_sock_container *sc, 370 struct delayed_work *work) 371{ 372 if (cancel_delayed_work(work)) 373 sc_put(sc); 374} 375 376static atomic_t o2net_connected_peers = ATOMIC_INIT(0); 377 378int o2net_num_connected_peers(void) 379{ 380 return atomic_read(&o2net_connected_peers); 381} 382 383static void o2net_set_nn_state(struct o2net_node *nn, 384 struct o2net_sock_container *sc, 385 unsigned valid, int err) 386{ 387 int was_valid = nn->nn_sc_valid; 388 int was_err = nn->nn_persistent_error; 389 struct o2net_sock_container *old_sc = nn->nn_sc; 390 391 assert_spin_locked(&nn->nn_lock); 392 393 if (old_sc && !sc) 394 atomic_dec(&o2net_connected_peers); 395 else if (!old_sc && sc) 396 atomic_inc(&o2net_connected_peers); 397 398 /* the node num comparison and single connect/accept path should stop 399 * an non-null sc from being overwritten with another */ 400 BUG_ON(sc && nn->nn_sc && nn->nn_sc != sc); 401 mlog_bug_on_msg(err && valid, "err %d valid %u\n", err, valid); 402 mlog_bug_on_msg(valid && !sc, "valid %u sc %p\n", valid, sc); 403 404 /* we won't reconnect after our valid conn goes away for 405 * this hb iteration.. here so it shows up in the logs */ 406 if (was_valid && !valid && err == 0) 407 err = -ENOTCONN; 408 409 mlog(ML_CONN, "node %u sc: %p -> %p, valid %u -> %u, err %d -> %d\n", 410 o2net_num_from_nn(nn), nn->nn_sc, sc, nn->nn_sc_valid, valid, 411 nn->nn_persistent_error, err); 412 413 nn->nn_sc = sc; 414 nn->nn_sc_valid = valid ? 1 : 0; 415 nn->nn_persistent_error = err; 416 417 /* mirrors o2net_tx_can_proceed() */ 418 if (nn->nn_persistent_error || nn->nn_sc_valid) 419 wake_up(&nn->nn_sc_wq); 420 421 if (!was_err && nn->nn_persistent_error) { 422 o2quo_conn_err(o2net_num_from_nn(nn)); 423 queue_delayed_work(o2net_wq, &nn->nn_still_up, 424 msecs_to_jiffies(O2NET_QUORUM_DELAY_MS)); 425 } 426 427 if (was_valid && !valid) { 428 printk(KERN_INFO "o2net: no longer connected to " 429 SC_NODEF_FMT "\n", SC_NODEF_ARGS(old_sc)); 430 o2net_complete_nodes_nsw(nn); 431 } 432 433 if (!was_valid && valid) { 434 o2quo_conn_up(o2net_num_from_nn(nn)); 435 /* this is a bit of a hack. we only try reconnecting 436 * when heartbeating starts until we get a connection. 437 * if that connection then dies we don't try reconnecting. 438 * the only way to start connecting again is to down 439 * heartbeat and bring it back up. */ 440 cancel_delayed_work(&nn->nn_connect_expired); 441 printk(KERN_INFO "o2net: %s " SC_NODEF_FMT "\n", 442 o2nm_this_node() > sc->sc_node->nd_num ? 443 "connected to" : "accepted connection from", 444 SC_NODEF_ARGS(sc)); 445 } 446 447 /* trigger the connecting worker func as long as we're not valid, 448 * it will back off if it shouldn't connect. This can be called 449 * from node config teardown and so needs to be careful about 450 * the work queue actually being up. */ 451 if (!valid && o2net_wq) { 452 unsigned long delay; 453 /* delay if we're withing a RECONNECT_DELAY of the 454 * last attempt */ 455 delay = (nn->nn_last_connect_attempt + 456 msecs_to_jiffies(o2net_reconnect_delay(sc->sc_node))) 457 - jiffies; 458 if (delay > msecs_to_jiffies(o2net_reconnect_delay(sc->sc_node))) 459 delay = 0; 460 mlog(ML_CONN, "queueing conn attempt in %lu jiffies\n", delay); 461 queue_delayed_work(o2net_wq, &nn->nn_connect_work, delay); 462 } 463 464 /* keep track of the nn's sc ref for the caller */ 465 if ((old_sc == NULL) && sc) 466 sc_get(sc); 467 if (old_sc && (old_sc != sc)) { 468 o2net_sc_queue_work(old_sc, &old_sc->sc_shutdown_work); 469 sc_put(old_sc); 470 } 471} 472 473/* see o2net_register_callbacks() */ 474static void o2net_data_ready(struct sock *sk, int bytes) 475{ 476 void (*ready)(struct sock *sk, int bytes); 477 478 read_lock(&sk->sk_callback_lock); 479 if (sk->sk_user_data) { 480 struct o2net_sock_container *sc = sk->sk_user_data; 481 sclog(sc, "data_ready hit\n"); 482 do_gettimeofday(&sc->sc_tv_data_ready); 483 o2net_sc_queue_work(sc, &sc->sc_rx_work); 484 ready = sc->sc_data_ready; 485 } else { 486 ready = sk->sk_data_ready; 487 } 488 read_unlock(&sk->sk_callback_lock); 489 490 ready(sk, bytes); 491} 492 493/* see o2net_register_callbacks() */ 494static void o2net_state_change(struct sock *sk) 495{ 496 void (*state_change)(struct sock *sk); 497 struct o2net_sock_container *sc; 498 499 read_lock(&sk->sk_callback_lock); 500 sc = sk->sk_user_data; 501 if (sc == NULL) { 502 state_change = sk->sk_state_change; 503 goto out; 504 } 505 506 sclog(sc, "state_change to %d\n", sk->sk_state); 507 508 state_change = sc->sc_state_change; 509 510 switch(sk->sk_state) { 511 /* ignore connecting sockets as they make progress */ 512 case TCP_SYN_SENT: 513 case TCP_SYN_RECV: 514 break; 515 case TCP_ESTABLISHED: 516 o2net_sc_queue_work(sc, &sc->sc_connect_work); 517 break; 518 default: 519 o2net_sc_queue_work(sc, &sc->sc_shutdown_work); 520 break; 521 } 522out: 523 read_unlock(&sk->sk_callback_lock); 524 state_change(sk); 525} 526 527/* 528 * we register callbacks so we can queue work on events before calling 529 * the original callbacks. our callbacks our careful to test user_data 530 * to discover when they've reaced with o2net_unregister_callbacks(). 531 */ 532static void o2net_register_callbacks(struct sock *sk, 533 struct o2net_sock_container *sc) 534{ 535 write_lock_bh(&sk->sk_callback_lock); 536 537 /* accepted sockets inherit the old listen socket data ready */ 538 if (sk->sk_data_ready == o2net_listen_data_ready) { 539 sk->sk_data_ready = sk->sk_user_data; 540 sk->sk_user_data = NULL; 541 } 542 543 BUG_ON(sk->sk_user_data != NULL); 544 sk->sk_user_data = sc; 545 sc_get(sc); 546 547 sc->sc_data_ready = sk->sk_data_ready; 548 sc->sc_state_change = sk->sk_state_change; 549 sk->sk_data_ready = o2net_data_ready; 550 sk->sk_state_change = o2net_state_change; 551 552 mutex_init(&sc->sc_send_lock); 553 554 write_unlock_bh(&sk->sk_callback_lock); 555} 556 557static int o2net_unregister_callbacks(struct sock *sk, 558 struct o2net_sock_container *sc) 559{ 560 int ret = 0; 561 562 write_lock_bh(&sk->sk_callback_lock); 563 if (sk->sk_user_data == sc) { 564 ret = 1; 565 sk->sk_user_data = NULL; 566 sk->sk_data_ready = sc->sc_data_ready; 567 sk->sk_state_change = sc->sc_state_change; 568 } 569 write_unlock_bh(&sk->sk_callback_lock); 570 571 return ret; 572} 573 574/* 575 * this is a little helper that is called by callers who have seen a problem 576 * with an sc and want to detach it from the nn if someone already hasn't beat 577 * them to it. if an error is given then the shutdown will be persistent 578 * and pending transmits will be canceled. 579 */ 580static void o2net_ensure_shutdown(struct o2net_node *nn, 581 struct o2net_sock_container *sc, 582 int err) 583{ 584 spin_lock(&nn->nn_lock); 585 if (nn->nn_sc == sc) 586 o2net_set_nn_state(nn, NULL, 0, err); 587 spin_unlock(&nn->nn_lock); 588} 589 590/* 591 * This work queue function performs the blocking parts of socket shutdown. A 592 * few paths lead here. set_nn_state will trigger this callback if it sees an 593 * sc detached from the nn. state_change will also trigger this callback 594 * directly when it sees errors. In that case we need to call set_nn_state 595 * ourselves as state_change couldn't get the nn_lock and call set_nn_state 596 * itself. 597 */ 598static void o2net_shutdown_sc(struct work_struct *work) 599{ 600 struct o2net_sock_container *sc = 601 container_of(work, struct o2net_sock_container, 602 sc_shutdown_work); 603 struct o2net_node *nn = o2net_nn_from_num(sc->sc_node->nd_num); 604 605 sclog(sc, "shutting down\n"); 606 607 /* drop the callbacks ref and call shutdown only once */ 608 if (o2net_unregister_callbacks(sc->sc_sock->sk, sc)) { 609 /* we shouldn't flush as we're in the thread, the 610 * races with pending sc work structs are harmless */ 611 del_timer_sync(&sc->sc_idle_timeout); 612 o2net_sc_cancel_delayed_work(sc, &sc->sc_keepalive_work); 613 sc_put(sc); 614 sc->sc_sock->ops->shutdown(sc->sc_sock, 615 RCV_SHUTDOWN|SEND_SHUTDOWN); 616 } 617 618 /* not fatal so failed connects before the other guy has our 619 * heartbeat can be retried */ 620 o2net_ensure_shutdown(nn, sc, 0); 621 sc_put(sc); 622} 623 624/* ------------------------------------------------------------ */ 625 626static int o2net_handler_cmp(struct o2net_msg_handler *nmh, u32 msg_type, 627 u32 key) 628{ 629 int ret = memcmp(&nmh->nh_key, &key, sizeof(key)); 630 631 if (ret == 0) 632 ret = memcmp(&nmh->nh_msg_type, &msg_type, sizeof(msg_type)); 633 634 return ret; 635} 636 637static struct o2net_msg_handler * 638o2net_handler_tree_lookup(u32 msg_type, u32 key, struct rb_node ***ret_p, 639 struct rb_node **ret_parent) 640{ 641 struct rb_node **p = &o2net_handler_tree.rb_node; 642 struct rb_node *parent = NULL; 643 struct o2net_msg_handler *nmh, *ret = NULL; 644 int cmp; 645 646 while (*p) { 647 parent = *p; 648 nmh = rb_entry(parent, struct o2net_msg_handler, nh_node); 649 cmp = o2net_handler_cmp(nmh, msg_type, key); 650 651 if (cmp < 0) 652 p = &(*p)->rb_left; 653 else if (cmp > 0) 654 p = &(*p)->rb_right; 655 else { 656 ret = nmh; 657 break; 658 } 659 } 660 661 if (ret_p != NULL) 662 *ret_p = p; 663 if (ret_parent != NULL) 664 *ret_parent = parent; 665 666 return ret; 667} 668 669static void o2net_handler_kref_release(struct kref *kref) 670{ 671 struct o2net_msg_handler *nmh; 672 nmh = container_of(kref, struct o2net_msg_handler, nh_kref); 673 674 kfree(nmh); 675} 676 677static void o2net_handler_put(struct o2net_msg_handler *nmh) 678{ 679 kref_put(&nmh->nh_kref, o2net_handler_kref_release); 680} 681 682/* max_len is protection for the handler func. incoming messages won't 683 * be given to the handler if their payload is longer than the max. */ 684int o2net_register_handler(u32 msg_type, u32 key, u32 max_len, 685 o2net_msg_handler_func *func, void *data, 686 o2net_post_msg_handler_func *post_func, 687 struct list_head *unreg_list) 688{ 689 struct o2net_msg_handler *nmh = NULL; 690 struct rb_node **p, *parent; 691 int ret = 0; 692 693 if (max_len > O2NET_MAX_PAYLOAD_BYTES) { 694 mlog(0, "max_len for message handler out of range: %u\n", 695 max_len); 696 ret = -EINVAL; 697 goto out; 698 } 699 700 if (!msg_type) { 701 mlog(0, "no message type provided: %u, %p\n", msg_type, func); 702 ret = -EINVAL; 703 goto out; 704 705 } 706 if (!func) { 707 mlog(0, "no message handler provided: %u, %p\n", 708 msg_type, func); 709 ret = -EINVAL; 710 goto out; 711 } 712 713 nmh = kzalloc(sizeof(struct o2net_msg_handler), GFP_NOFS); 714 if (nmh == NULL) { 715 ret = -ENOMEM; 716 goto out; 717 } 718 719 nmh->nh_func = func; 720 nmh->nh_func_data = data; 721 nmh->nh_post_func = post_func; 722 nmh->nh_msg_type = msg_type; 723 nmh->nh_max_len = max_len; 724 nmh->nh_key = key; 725 /* the tree and list get this ref.. they're both removed in 726 * unregister when this ref is dropped */ 727 kref_init(&nmh->nh_kref); 728 INIT_LIST_HEAD(&nmh->nh_unregister_item); 729 730 write_lock(&o2net_handler_lock); 731 if (o2net_handler_tree_lookup(msg_type, key, &p, &parent)) 732 ret = -EEXIST; 733 else { 734 rb_link_node(&nmh->nh_node, parent, p); 735 rb_insert_color(&nmh->nh_node, &o2net_handler_tree); 736 list_add_tail(&nmh->nh_unregister_item, unreg_list); 737 738 mlog(ML_TCP, "registered handler func %p type %u key %08x\n", 739 func, msg_type, key); 740 /* we've had some trouble with handlers seemingly vanishing. */ 741 mlog_bug_on_msg(o2net_handler_tree_lookup(msg_type, key, &p, 742 &parent) == NULL, 743 "couldn't find handler we *just* registerd " 744 "for type %u key %08x\n", msg_type, key); 745 } 746 write_unlock(&o2net_handler_lock); 747 if (ret) 748 goto out; 749 750out: 751 if (ret) 752 kfree(nmh); 753 754 return ret; 755} 756EXPORT_SYMBOL_GPL(o2net_register_handler); 757 758void o2net_unregister_handler_list(struct list_head *list) 759{ 760 struct list_head *pos, *n; 761 struct o2net_msg_handler *nmh; 762 763 write_lock(&o2net_handler_lock); 764 list_for_each_safe(pos, n, list) { 765 nmh = list_entry(pos, struct o2net_msg_handler, 766 nh_unregister_item); 767 mlog(ML_TCP, "unregistering handler func %p type %u key %08x\n", 768 nmh->nh_func, nmh->nh_msg_type, nmh->nh_key); 769 rb_erase(&nmh->nh_node, &o2net_handler_tree); 770 list_del_init(&nmh->nh_unregister_item); 771 kref_put(&nmh->nh_kref, o2net_handler_kref_release); 772 } 773 write_unlock(&o2net_handler_lock); 774} 775EXPORT_SYMBOL_GPL(o2net_unregister_handler_list); 776 777static struct o2net_msg_handler *o2net_handler_get(u32 msg_type, u32 key) 778{ 779 struct o2net_msg_handler *nmh; 780 781 read_lock(&o2net_handler_lock); 782 nmh = o2net_handler_tree_lookup(msg_type, key, NULL, NULL); 783 if (nmh) 784 kref_get(&nmh->nh_kref); 785 read_unlock(&o2net_handler_lock); 786 787 return nmh; 788} 789 790/* ------------------------------------------------------------ */ 791 792static int o2net_recv_tcp_msg(struct socket *sock, void *data, size_t len) 793{ 794 int ret; 795 mm_segment_t oldfs; 796 struct kvec vec = { 797 .iov_len = len, 798 .iov_base = data, 799 }; 800 struct msghdr msg = { 801 .msg_iovlen = 1, 802 .msg_iov = (struct iovec *)&vec, 803 .msg_flags = MSG_DONTWAIT, 804 }; 805 806 oldfs = get_fs(); 807 set_fs(get_ds()); 808 ret = sock_recvmsg(sock, &msg, len, msg.msg_flags); 809 set_fs(oldfs); 810 811 return ret; 812} 813 814static int o2net_send_tcp_msg(struct socket *sock, struct kvec *vec, 815 size_t veclen, size_t total) 816{ 817 int ret; 818 mm_segment_t oldfs; 819 struct msghdr msg = { 820 .msg_iov = (struct iovec *)vec, 821 .msg_iovlen = veclen, 822 }; 823 824 if (sock == NULL) { 825 ret = -EINVAL; 826 goto out; 827 } 828 829 oldfs = get_fs(); 830 set_fs(get_ds()); 831 ret = sock_sendmsg(sock, &msg, total); 832 set_fs(oldfs); 833 if (ret != total) { 834 mlog(ML_ERROR, "sendmsg returned %d instead of %zu\n", ret, 835 total); 836 if (ret >= 0) 837 ret = -EPIPE; /* should be smarter, I bet */ 838 goto out; 839 } 840 841 ret = 0; 842out: 843 if (ret < 0) 844 mlog(0, "returning error: %d\n", ret); 845 return ret; 846} 847 848static void o2net_sendpage(struct o2net_sock_container *sc, 849 void *kmalloced_virt, 850 size_t size) 851{ 852 struct o2net_node *nn = o2net_nn_from_num(sc->sc_node->nd_num); 853 ssize_t ret; 854 855 856 mutex_lock(&sc->sc_send_lock); 857 ret = sc->sc_sock->ops->sendpage(sc->sc_sock, 858 virt_to_page(kmalloced_virt), 859 (long)kmalloced_virt & ~PAGE_MASK, 860 size, MSG_DONTWAIT); 861 mutex_unlock(&sc->sc_send_lock); 862 if (ret != size) { 863 mlog(ML_ERROR, "sendpage of size %zu to " SC_NODEF_FMT 864 " failed with %zd\n", size, SC_NODEF_ARGS(sc), ret); 865 o2net_ensure_shutdown(nn, sc, 0); 866 } 867} 868 869static void o2net_init_msg(struct o2net_msg *msg, u16 data_len, u16 msg_type, u32 key) 870{ 871 memset(msg, 0, sizeof(struct o2net_msg)); 872 msg->magic = cpu_to_be16(O2NET_MSG_MAGIC); 873 msg->data_len = cpu_to_be16(data_len); 874 msg->msg_type = cpu_to_be16(msg_type); 875 msg->sys_status = cpu_to_be32(O2NET_ERR_NONE); 876 msg->status = 0; 877 msg->key = cpu_to_be32(key); 878} 879 880static int o2net_tx_can_proceed(struct o2net_node *nn, 881 struct o2net_sock_container **sc_ret, 882 int *error) 883{ 884 int ret = 0; 885 886 spin_lock(&nn->nn_lock); 887 if (nn->nn_persistent_error) { 888 ret = 1; 889 *sc_ret = NULL; 890 *error = nn->nn_persistent_error; 891 } else if (nn->nn_sc_valid) { 892 kref_get(&nn->nn_sc->sc_kref); 893 894 ret = 1; 895 *sc_ret = nn->nn_sc; 896 *error = 0; 897 } 898 spin_unlock(&nn->nn_lock); 899 900 return ret; 901} 902 903int o2net_send_message_vec(u32 msg_type, u32 key, struct kvec *caller_vec, 904 size_t caller_veclen, u8 target_node, int *status) 905{ 906 int ret, error = 0; 907 struct o2net_msg *msg = NULL; 908 size_t veclen, caller_bytes = 0; 909 struct kvec *vec = NULL; 910 struct o2net_sock_container *sc = NULL; 911 struct o2net_node *nn = o2net_nn_from_num(target_node); 912 struct o2net_status_wait nsw = { 913 .ns_node_item = LIST_HEAD_INIT(nsw.ns_node_item), 914 }; 915 916 if (o2net_wq == NULL) { 917 mlog(0, "attempt to tx without o2netd running\n"); 918 ret = -ESRCH; 919 goto out; 920 } 921 922 if (caller_veclen == 0) { 923 mlog(0, "bad kvec array length\n"); 924 ret = -EINVAL; 925 goto out; 926 } 927 928 caller_bytes = iov_length((struct iovec *)caller_vec, caller_veclen); 929 if (caller_bytes > O2NET_MAX_PAYLOAD_BYTES) { 930 mlog(0, "total payload len %zu too large\n", caller_bytes); 931 ret = -EINVAL; 932 goto out; 933 } 934 935 if (target_node == o2nm_this_node()) { 936 ret = -ELOOP; 937 goto out; 938 } 939 940 ret = wait_event_interruptible(nn->nn_sc_wq, 941 o2net_tx_can_proceed(nn, &sc, &error)); 942 if (!ret && error) 943 ret = error; 944 if (ret) 945 goto out; 946 947 veclen = caller_veclen + 1; 948 vec = kmalloc(sizeof(struct kvec) * veclen, GFP_ATOMIC); 949 if (vec == NULL) { 950 mlog(0, "failed to %zu element kvec!\n", veclen); 951 ret = -ENOMEM; 952 goto out; 953 } 954 955 msg = kmalloc(sizeof(struct o2net_msg), GFP_ATOMIC); 956 if (!msg) { 957 mlog(0, "failed to allocate a o2net_msg!\n"); 958 ret = -ENOMEM; 959 goto out; 960 } 961 962 o2net_init_msg(msg, caller_bytes, msg_type, key); 963 964 vec[0].iov_len = sizeof(struct o2net_msg); 965 vec[0].iov_base = msg; 966 memcpy(&vec[1], caller_vec, caller_veclen * sizeof(struct kvec)); 967 968 ret = o2net_prep_nsw(nn, &nsw); 969 if (ret) 970 goto out; 971 972 msg->msg_num = cpu_to_be32(nsw.ns_id); 973 974 /* finally, convert the message header to network byte-order 975 * and send */ 976 mutex_lock(&sc->sc_send_lock); 977 ret = o2net_send_tcp_msg(sc->sc_sock, vec, veclen, 978 sizeof(struct o2net_msg) + caller_bytes); 979 mutex_unlock(&sc->sc_send_lock); 980 msglog(msg, "sending returned %d\n", ret); 981 if (ret < 0) { 982 mlog(0, "error returned from o2net_send_tcp_msg=%d\n", ret); 983 goto out; 984 } 985 986 /* wait on other node's handler */ 987 wait_event(nsw.ns_wq, o2net_nsw_completed(nn, &nsw)); 988 989 /* Note that we avoid overwriting the callers status return 990 * variable if a system error was reported on the other 991 * side. Callers beware. */ 992 ret = o2net_sys_err_to_errno(nsw.ns_sys_status); 993 if (status && !ret) 994 *status = nsw.ns_status; 995 996 mlog(0, "woken, returning system status %d, user status %d\n", 997 ret, nsw.ns_status); 998out: 999 if (sc) 1000 sc_put(sc); 1001 if (vec) 1002 kfree(vec); 1003 if (msg) 1004 kfree(msg); 1005 o2net_complete_nsw(nn, &nsw, 0, 0, 0); 1006 return ret; 1007} 1008EXPORT_SYMBOL_GPL(o2net_send_message_vec); 1009 1010int o2net_send_message(u32 msg_type, u32 key, void *data, u32 len, 1011 u8 target_node, int *status) 1012{ 1013 struct kvec vec = { 1014 .iov_base = data, 1015 .iov_len = len, 1016 }; 1017 return o2net_send_message_vec(msg_type, key, &vec, 1, 1018 target_node, status); 1019} 1020EXPORT_SYMBOL_GPL(o2net_send_message); 1021 1022static int o2net_send_status_magic(struct socket *sock, struct o2net_msg *hdr, 1023 enum o2net_system_error syserr, int err) 1024{ 1025 struct kvec vec = { 1026 .iov_base = hdr, 1027 .iov_len = sizeof(struct o2net_msg), 1028 }; 1029 1030 BUG_ON(syserr >= O2NET_ERR_MAX); 1031 1032 /* leave other fields intact from the incoming message, msg_num 1033 * in particular */ 1034 hdr->sys_status = cpu_to_be32(syserr); 1035 hdr->status = cpu_to_be32(err); 1036 hdr->magic = cpu_to_be16(O2NET_MSG_STATUS_MAGIC); // twiddle the magic 1037 hdr->data_len = 0; 1038 1039 msglog(hdr, "about to send status magic %d\n", err); 1040 /* hdr has been in host byteorder this whole time */ 1041 return o2net_send_tcp_msg(sock, &vec, 1, sizeof(struct o2net_msg)); 1042} 1043 1044/* this returns -errno if the header was unknown or too large, etc. 1045 * after this is called the buffer us reused for the next message */ 1046static int o2net_process_message(struct o2net_sock_container *sc, 1047 struct o2net_msg *hdr) 1048{ 1049 struct o2net_node *nn = o2net_nn_from_num(sc->sc_node->nd_num); 1050 int ret = 0, handler_status; 1051 enum o2net_system_error syserr; 1052 struct o2net_msg_handler *nmh = NULL; 1053 void *ret_data = NULL; 1054 1055 msglog(hdr, "processing message\n"); 1056 1057 o2net_sc_postpone_idle(sc); 1058 1059 switch(be16_to_cpu(hdr->magic)) { 1060 case O2NET_MSG_STATUS_MAGIC: 1061 /* special type for returning message status */ 1062 o2net_complete_nsw(nn, NULL, 1063 be32_to_cpu(hdr->msg_num), 1064 be32_to_cpu(hdr->sys_status), 1065 be32_to_cpu(hdr->status)); 1066 goto out; 1067 case O2NET_MSG_KEEP_REQ_MAGIC: 1068 o2net_sendpage(sc, o2net_keep_resp, 1069 sizeof(*o2net_keep_resp)); 1070 goto out; 1071 case O2NET_MSG_KEEP_RESP_MAGIC: 1072 goto out; 1073 case O2NET_MSG_MAGIC: 1074 break; 1075 default: 1076 msglog(hdr, "bad magic\n"); 1077 ret = -EINVAL; 1078 goto out; 1079 break; 1080 } 1081 1082 /* find a handler for it */ 1083 handler_status = 0; 1084 nmh = o2net_handler_get(be16_to_cpu(hdr->msg_type), 1085 be32_to_cpu(hdr->key)); 1086 if (!nmh) { 1087 mlog(ML_TCP, "couldn't find handler for type %u key %08x\n", 1088 be16_to_cpu(hdr->msg_type), be32_to_cpu(hdr->key)); 1089 syserr = O2NET_ERR_NO_HNDLR; 1090 goto out_respond; 1091 } 1092 1093 syserr = O2NET_ERR_NONE; 1094 1095 if (be16_to_cpu(hdr->data_len) > nmh->nh_max_len) 1096 syserr = O2NET_ERR_OVERFLOW; 1097 1098 if (syserr != O2NET_ERR_NONE) 1099 goto out_respond; 1100 1101 do_gettimeofday(&sc->sc_tv_func_start); 1102 sc->sc_msg_key = be32_to_cpu(hdr->key); 1103 sc->sc_msg_type = be16_to_cpu(hdr->msg_type); 1104 handler_status = (nmh->nh_func)(hdr, sizeof(struct o2net_msg) + 1105 be16_to_cpu(hdr->data_len), 1106 nmh->nh_func_data, &ret_data); 1107 do_gettimeofday(&sc->sc_tv_func_stop); 1108 1109out_respond: 1110 /* this destroys the hdr, so don't use it after this */ 1111 mutex_lock(&sc->sc_send_lock); 1112 ret = o2net_send_status_magic(sc->sc_sock, hdr, syserr, 1113 handler_status); 1114 mutex_unlock(&sc->sc_send_lock); 1115 hdr = NULL; 1116 mlog(0, "sending handler status %d, syserr %d returned %d\n", 1117 handler_status, syserr, ret); 1118 1119 if (nmh) { 1120 BUG_ON(ret_data != NULL && nmh->nh_post_func == NULL); 1121 if (nmh->nh_post_func) 1122 (nmh->nh_post_func)(handler_status, nmh->nh_func_data, 1123 ret_data); 1124 } 1125 1126out: 1127 if (nmh) 1128 o2net_handler_put(nmh); 1129 return ret; 1130} 1131 1132static int o2net_check_handshake(struct o2net_sock_container *sc) 1133{ 1134 struct o2net_handshake *hand = page_address(sc->sc_page); 1135 struct o2net_node *nn = o2net_nn_from_num(sc->sc_node->nd_num); 1136 1137 if (hand->protocol_version != cpu_to_be64(O2NET_PROTOCOL_VERSION)) { 1138 mlog(ML_NOTICE, SC_NODEF_FMT " advertised net protocol " 1139 "version %llu but %llu is required, disconnecting\n", 1140 SC_NODEF_ARGS(sc), 1141 (unsigned long long)be64_to_cpu(hand->protocol_version), 1142 O2NET_PROTOCOL_VERSION); 1143 1144 /* don't bother reconnecting if its the wrong version. */ 1145 o2net_ensure_shutdown(nn, sc, -ENOTCONN); 1146 return -1; 1147 } 1148 1149 /* 1150 * Ensure timeouts are consistent with other nodes, otherwise 1151 * we can end up with one node thinking that the other must be down, 1152 * but isn't. This can ultimately cause corruption. 1153 */ 1154 if (be32_to_cpu(hand->o2net_idle_timeout_ms) != 1155 o2net_idle_timeout(sc->sc_node)) { 1156 mlog(ML_NOTICE, SC_NODEF_FMT " uses a network idle timeout of " 1157 "%u ms, but we use %u ms locally. disconnecting\n", 1158 SC_NODEF_ARGS(sc), 1159 be32_to_cpu(hand->o2net_idle_timeout_ms), 1160 o2net_idle_timeout(sc->sc_node)); 1161 o2net_ensure_shutdown(nn, sc, -ENOTCONN); 1162 return -1; 1163 } 1164 1165 if (be32_to_cpu(hand->o2net_keepalive_delay_ms) != 1166 o2net_keepalive_delay(sc->sc_node)) { 1167 mlog(ML_NOTICE, SC_NODEF_FMT " uses a keepalive delay of " 1168 "%u ms, but we use %u ms locally. disconnecting\n", 1169 SC_NODEF_ARGS(sc), 1170 be32_to_cpu(hand->o2net_keepalive_delay_ms), 1171 o2net_keepalive_delay(sc->sc_node)); 1172 o2net_ensure_shutdown(nn, sc, -ENOTCONN); 1173 return -1; 1174 } 1175 1176 if (be32_to_cpu(hand->o2hb_heartbeat_timeout_ms) != 1177 O2HB_MAX_WRITE_TIMEOUT_MS) { 1178 mlog(ML_NOTICE, SC_NODEF_FMT " uses a heartbeat timeout of " 1179 "%u ms, but we use %u ms locally. disconnecting\n", 1180 SC_NODEF_ARGS(sc), 1181 be32_to_cpu(hand->o2hb_heartbeat_timeout_ms), 1182 O2HB_MAX_WRITE_TIMEOUT_MS); 1183 o2net_ensure_shutdown(nn, sc, -ENOTCONN); 1184 return -1; 1185 } 1186 1187 sc->sc_handshake_ok = 1; 1188 1189 spin_lock(&nn->nn_lock); 1190 /* set valid and queue the idle timers only if it hasn't been 1191 * shut down already */ 1192 if (nn->nn_sc == sc) { 1193 o2net_sc_reset_idle_timer(sc); 1194 o2net_set_nn_state(nn, sc, 1, 0); 1195 } 1196 spin_unlock(&nn->nn_lock); 1197 1198 /* shift everything up as though it wasn't there */ 1199 sc->sc_page_off -= sizeof(struct o2net_handshake); 1200 if (sc->sc_page_off) 1201 memmove(hand, hand + 1, sc->sc_page_off); 1202 1203 return 0; 1204} 1205 1206/* this demuxes the queued rx bytes into header or payload bits and calls 1207 * handlers as each full message is read off the socket. it returns -error, 1208 * == 0 eof, or > 0 for progress made.*/ 1209static int o2net_advance_rx(struct o2net_sock_container *sc) 1210{ 1211 struct o2net_msg *hdr; 1212 int ret = 0; 1213 void *data; 1214 size_t datalen; 1215 1216 sclog(sc, "receiving\n"); 1217 do_gettimeofday(&sc->sc_tv_advance_start); 1218 1219 if (unlikely(sc->sc_handshake_ok == 0)) { 1220 if(sc->sc_page_off < sizeof(struct o2net_handshake)) { 1221 data = page_address(sc->sc_page) + sc->sc_page_off; 1222 datalen = sizeof(struct o2net_handshake) - sc->sc_page_off; 1223 ret = o2net_recv_tcp_msg(sc->sc_sock, data, datalen); 1224 if (ret > 0) 1225 sc->sc_page_off += ret; 1226 } 1227 1228 if (sc->sc_page_off == sizeof(struct o2net_handshake)) { 1229 o2net_check_handshake(sc); 1230 if (unlikely(sc->sc_handshake_ok == 0)) 1231 ret = -EPROTO; 1232 } 1233 goto out; 1234 } 1235 1236 /* do we need more header? */ 1237 if (sc->sc_page_off < sizeof(struct o2net_msg)) { 1238 data = page_address(sc->sc_page) + sc->sc_page_off; 1239 datalen = sizeof(struct o2net_msg) - sc->sc_page_off; 1240 ret = o2net_recv_tcp_msg(sc->sc_sock, data, datalen); 1241 if (ret > 0) { 1242 sc->sc_page_off += ret; 1243 /* only swab incoming here.. we can 1244 * only get here once as we cross from 1245 * being under to over */ 1246 if (sc->sc_page_off == sizeof(struct o2net_msg)) { 1247 hdr = page_address(sc->sc_page); 1248 if (be16_to_cpu(hdr->data_len) > 1249 O2NET_MAX_PAYLOAD_BYTES) 1250 ret = -EOVERFLOW; 1251 } 1252 } 1253 if (ret <= 0) 1254 goto out; 1255 } 1256 1257 if (sc->sc_page_off < sizeof(struct o2net_msg)) { 1258 /* oof, still don't have a header */ 1259 goto out; 1260 } 1261 1262 /* this was swabbed above when we first read it */ 1263 hdr = page_address(sc->sc_page); 1264 1265 msglog(hdr, "at page_off %zu\n", sc->sc_page_off); 1266 1267 /* do we need more payload? */ 1268 if (sc->sc_page_off - sizeof(struct o2net_msg) < be16_to_cpu(hdr->data_len)) { 1269 /* need more payload */ 1270 data = page_address(sc->sc_page) + sc->sc_page_off; 1271 datalen = (sizeof(struct o2net_msg) + be16_to_cpu(hdr->data_len)) - 1272 sc->sc_page_off; 1273 ret = o2net_recv_tcp_msg(sc->sc_sock, data, datalen); 1274 if (ret > 0) 1275 sc->sc_page_off += ret; 1276 if (ret <= 0) 1277 goto out; 1278 } 1279 1280 if (sc->sc_page_off - sizeof(struct o2net_msg) == be16_to_cpu(hdr->data_len)) { 1281 /* we can only get here once, the first time we read 1282 * the payload.. so set ret to progress if the handler 1283 * works out. after calling this the message is toast */ 1284 ret = o2net_process_message(sc, hdr); 1285 if (ret == 0) 1286 ret = 1; 1287 sc->sc_page_off = 0; 1288 } 1289 1290out: 1291 sclog(sc, "ret = %d\n", ret); 1292 do_gettimeofday(&sc->sc_tv_advance_stop); 1293 return ret; 1294} 1295 1296/* this work func is triggerd by data ready. it reads until it can read no 1297 * more. it interprets 0, eof, as fatal. if data_ready hits while we're doing 1298 * our work the work struct will be marked and we'll be called again. */ 1299static void o2net_rx_until_empty(struct work_struct *work) 1300{ 1301 struct o2net_sock_container *sc = 1302 container_of(work, struct o2net_sock_container, sc_rx_work); 1303 int ret; 1304 1305 do { 1306 ret = o2net_advance_rx(sc); 1307 } while (ret > 0); 1308 1309 if (ret <= 0 && ret != -EAGAIN) { 1310 struct o2net_node *nn = o2net_nn_from_num(sc->sc_node->nd_num); 1311 sclog(sc, "saw error %d, closing\n", ret); 1312 /* not permanent so read failed handshake can retry */ 1313 o2net_ensure_shutdown(nn, sc, 0); 1314 } 1315 1316 sc_put(sc); 1317} 1318 1319static int o2net_set_nodelay(struct socket *sock) 1320{ 1321 int ret, val = 1; 1322 mm_segment_t oldfs; 1323 1324 oldfs = get_fs(); 1325 set_fs(KERNEL_DS); 1326 1327 /* 1328 * Dear unsuspecting programmer, 1329 * 1330 * Don't use sock_setsockopt() for SOL_TCP. It doesn't check its level 1331 * argument and assumes SOL_SOCKET so, say, your TCP_NODELAY will 1332 * silently turn into SO_DEBUG. 1333 * 1334 * Yours, 1335 * Keeper of hilariously fragile interfaces. 1336 */ 1337 ret = sock->ops->setsockopt(sock, SOL_TCP, TCP_NODELAY, 1338 (char __user *)&val, sizeof(val)); 1339 1340 set_fs(oldfs); 1341 return ret; 1342} 1343 1344static void o2net_initialize_handshake(void) 1345{ 1346 o2net_hand->o2hb_heartbeat_timeout_ms = cpu_to_be32( 1347 O2HB_MAX_WRITE_TIMEOUT_MS); 1348 o2net_hand->o2net_idle_timeout_ms = cpu_to_be32( 1349 o2net_idle_timeout(NULL)); 1350 o2net_hand->o2net_keepalive_delay_ms = cpu_to_be32( 1351 o2net_keepalive_delay(NULL)); 1352 o2net_hand->o2net_reconnect_delay_ms = cpu_to_be32( 1353 o2net_reconnect_delay(NULL)); 1354} 1355 1356/* ------------------------------------------------------------ */ 1357 1358/* called when a connect completes and after a sock is accepted. the 1359 * rx path will see the response and mark the sc valid */ 1360static void o2net_sc_connect_completed(struct work_struct *work) 1361{ 1362 struct o2net_sock_container *sc = 1363 container_of(work, struct o2net_sock_container, 1364 sc_connect_work); 1365 1366 mlog(ML_MSG, "sc sending handshake with ver %llu id %llx\n", 1367 (unsigned long long)O2NET_PROTOCOL_VERSION, 1368 (unsigned long long)be64_to_cpu(o2net_hand->connector_id)); 1369 1370 o2net_initialize_handshake(); 1371 o2net_sendpage(sc, o2net_hand, sizeof(*o2net_hand)); 1372 sc_put(sc); 1373} 1374 1375/* this is called as a work_struct func. */ 1376static void o2net_sc_send_keep_req(struct work_struct *work) 1377{ 1378 struct o2net_sock_container *sc = 1379 container_of(work, struct o2net_sock_container, 1380 sc_keepalive_work.work); 1381 1382 o2net_sendpage(sc, o2net_keep_req, sizeof(*o2net_keep_req)); 1383 sc_put(sc); 1384} 1385 1386/* socket shutdown does a del_timer_sync against this as it tears down. 1387 * we can't start this timer until we've got to the point in sc buildup 1388 * where shutdown is going to be involved */ 1389static void o2net_idle_timer(unsigned long data) 1390{ 1391 struct o2net_sock_container *sc = (struct o2net_sock_container *)data; 1392 struct timeval now; 1393 1394 do_gettimeofday(&now); 1395 1396 printk(KERN_INFO "o2net: connection to " SC_NODEF_FMT " has been idle for %u.%u " 1397 "seconds, shutting it down.\n", SC_NODEF_ARGS(sc), 1398 o2net_idle_timeout(sc->sc_node) / 1000, 1399 o2net_idle_timeout(sc->sc_node) % 1000); 1400 mlog(ML_NOTICE, "here are some times that might help debug the " 1401 "situation: (tmr %ld.%ld now %ld.%ld dr %ld.%ld adv " 1402 "%ld.%ld:%ld.%ld func (%08x:%u) %ld.%ld:%ld.%ld)\n", 1403 sc->sc_tv_timer.tv_sec, (long) sc->sc_tv_timer.tv_usec, 1404 now.tv_sec, (long) now.tv_usec, 1405 sc->sc_tv_data_ready.tv_sec, (long) sc->sc_tv_data_ready.tv_usec, 1406 sc->sc_tv_advance_start.tv_sec, 1407 (long) sc->sc_tv_advance_start.tv_usec, 1408 sc->sc_tv_advance_stop.tv_sec, 1409 (long) sc->sc_tv_advance_stop.tv_usec, 1410 sc->sc_msg_key, sc->sc_msg_type, 1411 sc->sc_tv_func_start.tv_sec, (long) sc->sc_tv_func_start.tv_usec, 1412 sc->sc_tv_func_stop.tv_sec, (long) sc->sc_tv_func_stop.tv_usec); 1413 1414 o2net_sc_queue_work(sc, &sc->sc_shutdown_work); 1415} 1416 1417static void o2net_sc_reset_idle_timer(struct o2net_sock_container *sc) 1418{ 1419 o2net_sc_cancel_delayed_work(sc, &sc->sc_keepalive_work); 1420 o2net_sc_queue_delayed_work(sc, &sc->sc_keepalive_work, 1421 msecs_to_jiffies(o2net_keepalive_delay(sc->sc_node))); 1422 do_gettimeofday(&sc->sc_tv_timer); 1423 mod_timer(&sc->sc_idle_timeout, 1424 jiffies + msecs_to_jiffies(o2net_idle_timeout(sc->sc_node))); 1425} 1426 1427static void o2net_sc_postpone_idle(struct o2net_sock_container *sc) 1428{ 1429 /* Only push out an existing timer */ 1430 if (timer_pending(&sc->sc_idle_timeout)) 1431 o2net_sc_reset_idle_timer(sc); 1432} 1433 1434/* this work func is kicked whenever a path sets the nn state which doesn't 1435 * have valid set. This includes seeing hb come up, losing a connection, 1436 * having a connect attempt fail, etc. This centralizes the logic which decides 1437 * if a connect attempt should be made or if we should give up and all future 1438 * transmit attempts should fail */ 1439static void o2net_start_connect(struct work_struct *work) 1440{ 1441 struct o2net_node *nn = 1442 container_of(work, struct o2net_node, nn_connect_work.work); 1443 struct o2net_sock_container *sc = NULL; 1444 struct o2nm_node *node = NULL, *mynode = NULL; 1445 struct socket *sock = NULL; 1446 struct sockaddr_in myaddr = {0, }, remoteaddr = {0, }; 1447 int ret = 0, stop; 1448 1449 /* if we're greater we initiate tx, otherwise we accept */ 1450 if (o2nm_this_node() <= o2net_num_from_nn(nn)) 1451 goto out; 1452 1453 /* watch for racing with tearing a node down */ 1454 node = o2nm_get_node_by_num(o2net_num_from_nn(nn)); 1455 if (node == NULL) { 1456 ret = 0; 1457 goto out; 1458 } 1459 1460 mynode = o2nm_get_node_by_num(o2nm_this_node()); 1461 if (mynode == NULL) { 1462 ret = 0; 1463 goto out; 1464 } 1465 1466 spin_lock(&nn->nn_lock); 1467 /* see if we already have one pending or have given up */ 1468 stop = (nn->nn_sc || nn->nn_persistent_error); 1469 spin_unlock(&nn->nn_lock); 1470 if (stop) 1471 goto out; 1472 1473 nn->nn_last_connect_attempt = jiffies; 1474 1475 sc = sc_alloc(node); 1476 if (sc == NULL) { 1477 mlog(0, "couldn't allocate sc\n"); 1478 ret = -ENOMEM; 1479 goto out; 1480 } 1481 1482 ret = sock_create(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock); 1483 if (ret < 0) { 1484 mlog(0, "can't create socket: %d\n", ret); 1485 goto out; 1486 } 1487 sc->sc_sock = sock; /* freed by sc_kref_release */ 1488 1489 sock->sk->sk_allocation = GFP_ATOMIC; 1490 1491 myaddr.sin_family = AF_INET; 1492 myaddr.sin_addr.s_addr = mynode->nd_ipv4_address; 1493 myaddr.sin_port = (__force u16)htons(0); /* any port */ 1494 1495 ret = sock->ops->bind(sock, (struct sockaddr *)&myaddr, 1496 sizeof(myaddr)); 1497 if (ret) { 1498 mlog(ML_ERROR, "bind failed with %d at address %u.%u.%u.%u\n", 1499 ret, NIPQUAD(mynode->nd_ipv4_address)); 1500 goto out; 1501 } 1502 1503 ret = o2net_set_nodelay(sc->sc_sock); 1504 if (ret) { 1505 mlog(ML_ERROR, "setting TCP_NODELAY failed with %d\n", ret); 1506 goto out; 1507 } 1508 1509 o2net_register_callbacks(sc->sc_sock->sk, sc); 1510 1511 spin_lock(&nn->nn_lock); 1512 /* handshake completion will set nn->nn_sc_valid */ 1513 o2net_set_nn_state(nn, sc, 0, 0); 1514 spin_unlock(&nn->nn_lock); 1515 1516 remoteaddr.sin_family = AF_INET; 1517 remoteaddr.sin_addr.s_addr = node->nd_ipv4_address; 1518 remoteaddr.sin_port = node->nd_ipv4_port; 1519 1520 ret = sc->sc_sock->ops->connect(sc->sc_sock, 1521 (struct sockaddr *)&remoteaddr, 1522 sizeof(remoteaddr), 1523 O_NONBLOCK); 1524 if (ret == -EINPROGRESS) 1525 ret = 0; 1526 1527out: 1528 if (ret) { 1529 mlog(ML_NOTICE, "connect attempt to " SC_NODEF_FMT " failed " 1530 "with errno %d\n", SC_NODEF_ARGS(sc), ret); 1531 /* 0 err so that another will be queued and attempted 1532 * from set_nn_state */ 1533 if (sc) 1534 o2net_ensure_shutdown(nn, sc, 0); 1535 } 1536 if (sc) 1537 sc_put(sc); 1538 if (node) 1539 o2nm_node_put(node); 1540 if (mynode) 1541 o2nm_node_put(mynode); 1542 1543 return; 1544} 1545 1546static void o2net_connect_expired(struct work_struct *work) 1547{ 1548 struct o2net_node *nn = 1549 container_of(work, struct o2net_node, nn_connect_expired.work); 1550 1551 spin_lock(&nn->nn_lock); 1552 if (!nn->nn_sc_valid) { 1553 struct o2nm_node *node = nn->nn_sc->sc_node; 1554 mlog(ML_ERROR, "no connection established with node %u after " 1555 "%u.%u seconds, giving up and returning errors.\n", 1556 o2net_num_from_nn(nn), 1557 o2net_idle_timeout(node) / 1000, 1558 o2net_idle_timeout(node) % 1000); 1559 1560 o2net_set_nn_state(nn, NULL, 0, -ENOTCONN); 1561 } 1562 spin_unlock(&nn->nn_lock); 1563} 1564 1565static void o2net_still_up(struct work_struct *work) 1566{ 1567 struct o2net_node *nn = 1568 container_of(work, struct o2net_node, nn_still_up.work); 1569 1570 o2quo_hb_still_up(o2net_num_from_nn(nn)); 1571} 1572 1573/* ------------------------------------------------------------ */ 1574 1575void o2net_disconnect_node(struct o2nm_node *node) 1576{ 1577 struct o2net_node *nn = o2net_nn_from_num(node->nd_num); 1578 1579 /* don't reconnect until it's heartbeating again */ 1580 spin_lock(&nn->nn_lock); 1581 o2net_set_nn_state(nn, NULL, 0, -ENOTCONN); 1582 spin_unlock(&nn->nn_lock); 1583 1584 if (o2net_wq) { 1585 cancel_delayed_work(&nn->nn_connect_expired); 1586 cancel_delayed_work(&nn->nn_connect_work); 1587 cancel_delayed_work(&nn->nn_still_up); 1588 flush_workqueue(o2net_wq); 1589 } 1590} 1591 1592static void o2net_hb_node_down_cb(struct o2nm_node *node, int node_num, 1593 void *data) 1594{ 1595 o2quo_hb_down(node_num); 1596 1597 if (node_num != o2nm_this_node()) 1598 o2net_disconnect_node(node); 1599 1600 BUG_ON(atomic_read(&o2net_connected_peers) < 0); 1601} 1602 1603static void o2net_hb_node_up_cb(struct o2nm_node *node, int node_num, 1604 void *data) 1605{ 1606 struct o2net_node *nn = o2net_nn_from_num(node_num); 1607 1608 o2quo_hb_up(node_num); 1609 1610 /* ensure an immediate connect attempt */ 1611 nn->nn_last_connect_attempt = jiffies - 1612 (msecs_to_jiffies(o2net_reconnect_delay(node)) + 1); 1613 1614 if (node_num != o2nm_this_node()) { 1615 /* heartbeat doesn't work unless a local node number is 1616 * configured and doing so brings up the o2net_wq, so we can 1617 * use it.. */ 1618 queue_delayed_work(o2net_wq, &nn->nn_connect_expired, 1619 msecs_to_jiffies(o2net_idle_timeout(node))); 1620 1621 /* believe it or not, accept and node hearbeating testing 1622 * can succeed for this node before we got here.. so 1623 * only use set_nn_state to clear the persistent error 1624 * if that hasn't already happened */ 1625 spin_lock(&nn->nn_lock); 1626 if (nn->nn_persistent_error) 1627 o2net_set_nn_state(nn, NULL, 0, 0); 1628 spin_unlock(&nn->nn_lock); 1629 } 1630} 1631 1632void o2net_unregister_hb_callbacks(void) 1633{ 1634 o2hb_unregister_callback(&o2net_hb_up); 1635 o2hb_unregister_callback(&o2net_hb_down); 1636} 1637 1638int o2net_register_hb_callbacks(void) 1639{ 1640 int ret; 1641 1642 o2hb_setup_callback(&o2net_hb_down, O2HB_NODE_DOWN_CB, 1643 o2net_hb_node_down_cb, NULL, O2NET_HB_PRI); 1644 o2hb_setup_callback(&o2net_hb_up, O2HB_NODE_UP_CB, 1645 o2net_hb_node_up_cb, NULL, O2NET_HB_PRI); 1646 1647 ret = o2hb_register_callback(&o2net_hb_up); 1648 if (ret == 0) 1649 ret = o2hb_register_callback(&o2net_hb_down); 1650 1651 if (ret) 1652 o2net_unregister_hb_callbacks(); 1653 1654 return ret; 1655} 1656 1657/* ------------------------------------------------------------ */ 1658 1659static int o2net_accept_one(struct socket *sock) 1660{ 1661 int ret, slen; 1662 struct sockaddr_in sin; 1663 struct socket *new_sock = NULL; 1664 struct o2nm_node *node = NULL; 1665 struct o2net_sock_container *sc = NULL; 1666 struct o2net_node *nn; 1667 1668 BUG_ON(sock == NULL); 1669 ret = sock_create_lite(sock->sk->sk_family, sock->sk->sk_type, 1670 sock->sk->sk_protocol, &new_sock); 1671 if (ret) 1672 goto out; 1673 1674 new_sock->type = sock->type; 1675 new_sock->ops = sock->ops; 1676 ret = sock->ops->accept(sock, new_sock, O_NONBLOCK); 1677 if (ret < 0) 1678 goto out; 1679 1680 new_sock->sk->sk_allocation = GFP_ATOMIC; 1681 1682 ret = o2net_set_nodelay(new_sock); 1683 if (ret) { 1684 mlog(ML_ERROR, "setting TCP_NODELAY failed with %d\n", ret); 1685 goto out; 1686 } 1687 1688 slen = sizeof(sin); 1689 ret = new_sock->ops->getname(new_sock, (struct sockaddr *) &sin, 1690 &slen, 1); 1691 if (ret < 0) 1692 goto out; 1693 1694 node = o2nm_get_node_by_ip((__force __be32)sin.sin_addr.s_addr); 1695 if (node == NULL) { 1696 mlog(ML_NOTICE, "attempt to connect from unknown node at " 1697 "%u.%u.%u.%u:%d\n", NIPQUAD(sin.sin_addr.s_addr), 1698 ntohs((__force __be16)sin.sin_port)); 1699 ret = -EINVAL; 1700 goto out; 1701 } 1702 1703 if (o2nm_this_node() > node->nd_num) { 1704 mlog(ML_NOTICE, "unexpected connect attempted from a lower " 1705 "numbered node '%s' at " "%u.%u.%u.%u:%d with num %u\n", 1706 node->nd_name, NIPQUAD(sin.sin_addr.s_addr), 1707 ntohs((__force __be16)sin.sin_port), node->nd_num); 1708 ret = -EINVAL; 1709 goto out; 1710 } 1711 1712 /* this happens all the time when the other node sees our heartbeat 1713 * and tries to connect before we see their heartbeat */ 1714 if (!o2hb_check_node_heartbeating_from_callback(node->nd_num)) { 1715 mlog(ML_CONN, "attempt to connect from node '%s' at " 1716 "%u.%u.%u.%u:%d but it isn't heartbeating\n", 1717 node->nd_name, NIPQUAD(sin.sin_addr.s_addr), 1718 ntohs((__force __be16)sin.sin_port)); 1719 ret = -EINVAL; 1720 goto out; 1721 } 1722 1723 nn = o2net_nn_from_num(node->nd_num); 1724 1725 spin_lock(&nn->nn_lock); 1726 if (nn->nn_sc) 1727 ret = -EBUSY; 1728 else 1729 ret = 0; 1730 spin_unlock(&nn->nn_lock); 1731 if (ret) { 1732 mlog(ML_NOTICE, "attempt to connect from node '%s' at " 1733 "%u.%u.%u.%u:%d but it already has an open connection\n", 1734 node->nd_name, NIPQUAD(sin.sin_addr.s_addr), 1735 ntohs((__force __be16)sin.sin_port)); 1736 goto out; 1737 } 1738 1739 sc = sc_alloc(node); 1740 if (sc == NULL) { 1741 ret = -ENOMEM; 1742 goto out; 1743 } 1744 1745 sc->sc_sock = new_sock; 1746 new_sock = NULL; 1747 1748 spin_lock(&nn->nn_lock); 1749 o2net_set_nn_state(nn, sc, 0, 0); 1750 spin_unlock(&nn->nn_lock); 1751 1752 o2net_register_callbacks(sc->sc_sock->sk, sc); 1753 o2net_sc_queue_work(sc, &sc->sc_rx_work); 1754 1755 o2net_initialize_handshake(); 1756 o2net_sendpage(sc, o2net_hand, sizeof(*o2net_hand)); 1757 1758out: 1759 if (new_sock) 1760 sock_release(new_sock); 1761 if (node) 1762 o2nm_node_put(node); 1763 if (sc) 1764 sc_put(sc); 1765 return ret; 1766} 1767 1768static void o2net_accept_many(struct work_struct *work) 1769{ 1770 struct socket *sock = o2net_listen_sock; 1771 while (o2net_accept_one(sock) == 0) 1772 cond_resched(); 1773} 1774 1775static void o2net_listen_data_ready(struct sock *sk, int bytes) 1776{ 1777 void (*ready)(struct sock *sk, int bytes); 1778 1779 read_lock(&sk->sk_callback_lock); 1780 ready = sk->sk_user_data; 1781 if (ready == NULL) { /* check for teardown race */ 1782 ready = sk->sk_data_ready; 1783 goto out; 1784 } 1785 1786 /* ->sk_data_ready is also called for a newly established child socket 1787 * before it has been accepted and the acceptor has set up their 1788 * data_ready.. we only want to queue listen work for our listening 1789 * socket */ 1790 if (sk->sk_state == TCP_LISTEN) { 1791 mlog(ML_TCP, "bytes: %d\n", bytes); 1792 queue_work(o2net_wq, &o2net_listen_work); 1793 } 1794 1795out: 1796 read_unlock(&sk->sk_callback_lock); 1797 ready(sk, bytes); 1798} 1799 1800static int o2net_open_listening_sock(__be32 addr, __be16 port) 1801{ 1802 struct socket *sock = NULL; 1803 int ret; 1804 struct sockaddr_in sin = { 1805 .sin_family = PF_INET, 1806 .sin_addr = { .s_addr = addr }, 1807 .sin_port = port, 1808 }; 1809 1810 ret = sock_create(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock); 1811 if (ret < 0) { 1812 mlog(ML_ERROR, "unable to create socket, ret=%d\n", ret); 1813 goto out; 1814 } 1815 1816 sock->sk->sk_allocation = GFP_ATOMIC; 1817 1818 write_lock_bh(&sock->sk->sk_callback_lock); 1819 sock->sk->sk_user_data = sock->sk->sk_data_ready; 1820 sock->sk->sk_data_ready = o2net_listen_data_ready; 1821 write_unlock_bh(&sock->sk->sk_callback_lock); 1822 1823 o2net_listen_sock = sock; 1824 INIT_WORK(&o2net_listen_work, o2net_accept_many); 1825 1826 sock->sk->sk_reuse = 1; 1827 ret = sock->ops->bind(sock, (struct sockaddr *)&sin, sizeof(sin)); 1828 if (ret < 0) { 1829 mlog(ML_ERROR, "unable to bind socket at %u.%u.%u.%u:%u, " 1830 "ret=%d\n", NIPQUAD(addr), ntohs(port), ret); 1831 goto out; 1832 } 1833 1834 ret = sock->ops->listen(sock, 64); 1835 if (ret < 0) { 1836 mlog(ML_ERROR, "unable to listen on %u.%u.%u.%u:%u, ret=%d\n", 1837 NIPQUAD(addr), ntohs(port), ret); 1838 } 1839 1840out: 1841 if (ret) { 1842 o2net_listen_sock = NULL; 1843 if (sock) 1844 sock_release(sock); 1845 } 1846 return ret; 1847} 1848 1849/* 1850 * called from node manager when we should bring up our network listening 1851 * socket. node manager handles all the serialization to only call this 1852 * once and to match it with o2net_stop_listening(). note, 1853 * o2nm_this_node() doesn't work yet as we're being called while it 1854 * is being set up. 1855 */ 1856int o2net_start_listening(struct o2nm_node *node) 1857{ 1858 int ret = 0; 1859 1860 BUG_ON(o2net_wq != NULL); 1861 BUG_ON(o2net_listen_sock != NULL); 1862 1863 mlog(ML_KTHREAD, "starting o2net thread...\n"); 1864 o2net_wq = create_singlethread_workqueue("o2net"); 1865 if (o2net_wq == NULL) { 1866 mlog(ML_ERROR, "unable to launch o2net thread\n"); 1867 return -ENOMEM; /* ? */ 1868 } 1869 1870 ret = o2net_open_listening_sock(node->nd_ipv4_address, 1871 node->nd_ipv4_port); 1872 if (ret) { 1873 destroy_workqueue(o2net_wq); 1874 o2net_wq = NULL; 1875 } else 1876 o2quo_conn_up(node->nd_num); 1877 1878 return ret; 1879} 1880 1881/* again, o2nm_this_node() doesn't work here as we're involved in 1882 * tearing it down */ 1883void o2net_stop_listening(struct o2nm_node *node) 1884{ 1885 struct socket *sock = o2net_listen_sock; 1886 size_t i; 1887 1888 BUG_ON(o2net_wq == NULL); 1889 BUG_ON(o2net_listen_sock == NULL); 1890 1891 /* stop the listening socket from generating work */ 1892 write_lock_bh(&sock->sk->sk_callback_lock); 1893 sock->sk->sk_data_ready = sock->sk->sk_user_data; 1894 sock->sk->sk_user_data = NULL; 1895 write_unlock_bh(&sock->sk->sk_callback_lock); 1896 1897 for (i = 0; i < ARRAY_SIZE(o2net_nodes); i++) { 1898 struct o2nm_node *node = o2nm_get_node_by_num(i); 1899 if (node) { 1900 o2net_disconnect_node(node); 1901 o2nm_node_put(node); 1902 } 1903 } 1904 1905 /* finish all work and tear down the work queue */ 1906 mlog(ML_KTHREAD, "waiting for o2net thread to exit....\n"); 1907 destroy_workqueue(o2net_wq); 1908 o2net_wq = NULL; 1909 1910 sock_release(o2net_listen_sock); 1911 o2net_listen_sock = NULL; 1912 1913 o2quo_conn_err(node->nd_num); 1914} 1915 1916/* ------------------------------------------------------------ */ 1917 1918int o2net_init(void) 1919{ 1920 unsigned long i; 1921 1922 o2quo_init(); 1923 1924 o2net_hand = kzalloc(sizeof(struct o2net_handshake), GFP_KERNEL); 1925 o2net_keep_req = kzalloc(sizeof(struct o2net_msg), GFP_KERNEL); 1926 o2net_keep_resp = kzalloc(sizeof(struct o2net_msg), GFP_KERNEL); 1927 if (!o2net_hand || !o2net_keep_req || !o2net_keep_resp) { 1928 kfree(o2net_hand); 1929 kfree(o2net_keep_req); 1930 kfree(o2net_keep_resp); 1931 return -ENOMEM; 1932 } 1933 1934 o2net_hand->protocol_version = cpu_to_be64(O2NET_PROTOCOL_VERSION); 1935 o2net_hand->connector_id = cpu_to_be64(1); 1936 1937 o2net_keep_req->magic = cpu_to_be16(O2NET_MSG_KEEP_REQ_MAGIC); 1938 o2net_keep_resp->magic = cpu_to_be16(O2NET_MSG_KEEP_RESP_MAGIC); 1939 1940 for (i = 0; i < ARRAY_SIZE(o2net_nodes); i++) { 1941 struct o2net_node *nn = o2net_nn_from_num(i); 1942 1943 spin_lock_init(&nn->nn_lock); 1944 INIT_DELAYED_WORK(&nn->nn_connect_work, o2net_start_connect); 1945 INIT_DELAYED_WORK(&nn->nn_connect_expired, 1946 o2net_connect_expired); 1947 INIT_DELAYED_WORK(&nn->nn_still_up, o2net_still_up); 1948 /* until we see hb from a node we'll return einval */ 1949 nn->nn_persistent_error = -ENOTCONN; 1950 init_waitqueue_head(&nn->nn_sc_wq); 1951 idr_init(&nn->nn_status_idr); 1952 INIT_LIST_HEAD(&nn->nn_status_list); 1953 } 1954 1955 return 0; 1956} 1957 1958void o2net_exit(void) 1959{ 1960 o2quo_exit(); 1961 kfree(o2net_hand); 1962 kfree(o2net_keep_req); 1963 kfree(o2net_keep_resp); 1964} 1965