1 /* -*- mode: c; c-basic-offset: 8; -*-
3 * vim: noexpandtab sw=8 ts=8 sts=0:
5 * Copyright (C) 2004 Oracle. All rights reserved.
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.
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.
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.
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.
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
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
41 * Any framing errors (bad magic, large payload lengths) close a connection.
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.
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.
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>
63 #include <asm/uaccess.h>
65 #include "heartbeat.h"
67 #include "nodemanager.h"
68 #define MLOG_MASK_PREFIX ML_TCP
72 #include "tcp_internal.h"
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.
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)
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.
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); \
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 , \
111 static DEFINE_RWLOCK(o2net_handler_lock
);
112 static struct rb_root o2net_handler_tree
= RB_ROOT
;
114 static struct o2net_node o2net_nodes
[O2NM_MAX_NODES
];
116 /* XXX someday we'll need better accounting */
117 static struct socket
*o2net_listen_sock
= NULL
;
120 * listen work is only queued by the listening socket callbacks on the
121 * o2net_wq. teardown detaches the callbacks before destroying the workqueue.
122 * quorum work is queued as sock containers are shutdown.. stop_listening
123 * tears down all the node's sock containers, preventing future shutdowns
124 * and queued quroum work, before canceling delayed quorum work and
125 * destroying the work queue.
127 static struct workqueue_struct
*o2net_wq
;
128 static struct work_struct o2net_listen_work
;
130 static struct o2hb_callback_func o2net_hb_up
, o2net_hb_down
;
131 #define O2NET_HB_PRI 0x1
133 static struct o2net_handshake
*o2net_hand
;
134 static struct o2net_msg
*o2net_keep_req
, *o2net_keep_resp
;
136 static int o2net_sys_err_translations
[O2NET_ERR_MAX
] =
137 {[O2NET_ERR_NONE
] = 0,
138 [O2NET_ERR_NO_HNDLR
] = -ENOPROTOOPT
,
139 [O2NET_ERR_OVERFLOW
] = -EOVERFLOW
,
140 [O2NET_ERR_DIED
] = -EHOSTDOWN
,};
142 /* can't quite avoid *all* internal declarations :/ */
143 static void o2net_sc_connect_completed(struct work_struct
*work
);
144 static void o2net_rx_until_empty(struct work_struct
*work
);
145 static void o2net_shutdown_sc(struct work_struct
*work
);
146 static void o2net_listen_data_ready(struct sock
*sk
, int bytes
);
147 static void o2net_sc_send_keep_req(struct work_struct
*work
);
148 static void o2net_idle_timer(unsigned long data
);
149 static void o2net_sc_postpone_idle(struct o2net_sock_container
*sc
);
151 static inline int o2net_sys_err_to_errno(enum o2net_system_error err
)
154 BUG_ON(err
>= O2NET_ERR_MAX
);
155 trans
= o2net_sys_err_translations
[err
];
157 /* Just in case we mess up the translation table above */
158 BUG_ON(err
!= O2NET_ERR_NONE
&& trans
== 0);
162 static struct o2net_node
* o2net_nn_from_num(u8 node_num
)
164 BUG_ON(node_num
>= ARRAY_SIZE(o2net_nodes
));
165 return &o2net_nodes
[node_num
];
168 static u8
o2net_num_from_nn(struct o2net_node
*nn
)
171 return nn
- o2net_nodes
;
174 /* ------------------------------------------------------------ */
176 static int o2net_prep_nsw(struct o2net_node
*nn
, struct o2net_status_wait
*nsw
)
181 if (!idr_pre_get(&nn
->nn_status_idr
, GFP_ATOMIC
)) {
185 spin_lock(&nn
->nn_lock
);
186 ret
= idr_get_new(&nn
->nn_status_idr
, nsw
, &nsw
->ns_id
);
188 list_add_tail(&nsw
->ns_node_item
,
189 &nn
->nn_status_list
);
190 spin_unlock(&nn
->nn_lock
);
191 } while (ret
== -EAGAIN
);
194 init_waitqueue_head(&nsw
->ns_wq
);
195 nsw
->ns_sys_status
= O2NET_ERR_NONE
;
202 static void o2net_complete_nsw_locked(struct o2net_node
*nn
,
203 struct o2net_status_wait
*nsw
,
204 enum o2net_system_error sys_status
,
207 assert_spin_locked(&nn
->nn_lock
);
209 if (!list_empty(&nsw
->ns_node_item
)) {
210 list_del_init(&nsw
->ns_node_item
);
211 nsw
->ns_sys_status
= sys_status
;
212 nsw
->ns_status
= status
;
213 idr_remove(&nn
->nn_status_idr
, nsw
->ns_id
);
214 wake_up(&nsw
->ns_wq
);
218 static void o2net_complete_nsw(struct o2net_node
*nn
,
219 struct o2net_status_wait
*nsw
,
220 u64 id
, enum o2net_system_error sys_status
,
223 spin_lock(&nn
->nn_lock
);
228 nsw
= idr_find(&nn
->nn_status_idr
, id
);
233 o2net_complete_nsw_locked(nn
, nsw
, sys_status
, status
);
236 spin_unlock(&nn
->nn_lock
);
240 static void o2net_complete_nodes_nsw(struct o2net_node
*nn
)
242 struct list_head
*iter
, *tmp
;
243 unsigned int num_kills
= 0;
244 struct o2net_status_wait
*nsw
;
246 assert_spin_locked(&nn
->nn_lock
);
248 list_for_each_safe(iter
, tmp
, &nn
->nn_status_list
) {
249 nsw
= list_entry(iter
, struct o2net_status_wait
, ns_node_item
);
250 o2net_complete_nsw_locked(nn
, nsw
, O2NET_ERR_DIED
, 0);
254 mlog(0, "completed %d messages for node %u\n", num_kills
,
255 o2net_num_from_nn(nn
));
258 static int o2net_nsw_completed(struct o2net_node
*nn
,
259 struct o2net_status_wait
*nsw
)
262 spin_lock(&nn
->nn_lock
);
263 completed
= list_empty(&nsw
->ns_node_item
);
264 spin_unlock(&nn
->nn_lock
);
268 /* ------------------------------------------------------------ */
270 static void sc_kref_release(struct kref
*kref
)
272 struct o2net_sock_container
*sc
= container_of(kref
,
273 struct o2net_sock_container
, sc_kref
);
274 sclog(sc
, "releasing\n");
277 sock_release(sc
->sc_sock
);
281 o2nm_node_put(sc
->sc_node
);
287 static void sc_put(struct o2net_sock_container
*sc
)
290 kref_put(&sc
->sc_kref
, sc_kref_release
);
292 static void sc_get(struct o2net_sock_container
*sc
)
295 kref_get(&sc
->sc_kref
);
297 static struct o2net_sock_container
*sc_alloc(struct o2nm_node
*node
)
299 struct o2net_sock_container
*sc
, *ret
= NULL
;
300 struct page
*page
= NULL
;
302 page
= alloc_page(GFP_NOFS
);
303 sc
= kcalloc(1, sizeof(*sc
), GFP_NOFS
);
304 if (sc
== NULL
|| page
== NULL
)
307 kref_init(&sc
->sc_kref
);
311 INIT_WORK(&sc
->sc_connect_work
, o2net_sc_connect_completed
);
312 INIT_WORK(&sc
->sc_rx_work
, o2net_rx_until_empty
);
313 INIT_WORK(&sc
->sc_shutdown_work
, o2net_shutdown_sc
);
314 INIT_DELAYED_WORK(&sc
->sc_keepalive_work
, o2net_sc_send_keep_req
);
316 init_timer(&sc
->sc_idle_timeout
);
317 sc
->sc_idle_timeout
.function
= o2net_idle_timer
;
318 sc
->sc_idle_timeout
.data
= (unsigned long)sc
;
320 sclog(sc
, "alloced\n");
335 /* ------------------------------------------------------------ */
337 static void o2net_sc_queue_work(struct o2net_sock_container
*sc
,
338 struct work_struct
*work
)
341 if (!queue_work(o2net_wq
, work
))
344 static void o2net_sc_queue_delayed_work(struct o2net_sock_container
*sc
,
345 struct delayed_work
*work
,
349 if (!queue_delayed_work(o2net_wq
, work
, delay
))
352 static void o2net_sc_cancel_delayed_work(struct o2net_sock_container
*sc
,
353 struct delayed_work
*work
)
355 if (cancel_delayed_work(work
))
359 static void o2net_set_nn_state(struct o2net_node
*nn
,
360 struct o2net_sock_container
*sc
,
361 unsigned valid
, int err
)
363 int was_valid
= nn
->nn_sc_valid
;
364 int was_err
= nn
->nn_persistent_error
;
365 struct o2net_sock_container
*old_sc
= nn
->nn_sc
;
367 assert_spin_locked(&nn
->nn_lock
);
369 /* the node num comparison and single connect/accept path should stop
370 * an non-null sc from being overwritten with another */
371 BUG_ON(sc
&& nn
->nn_sc
&& nn
->nn_sc
!= sc
);
372 mlog_bug_on_msg(err
&& valid
, "err %d valid %u\n", err
, valid
);
373 mlog_bug_on_msg(valid
&& !sc
, "valid %u sc %p\n", valid
, sc
);
375 /* we won't reconnect after our valid conn goes away for
376 * this hb iteration.. here so it shows up in the logs */
377 if (was_valid
&& !valid
&& err
== 0)
380 mlog(ML_CONN
, "node %u sc: %p -> %p, valid %u -> %u, err %d -> %d\n",
381 o2net_num_from_nn(nn
), nn
->nn_sc
, sc
, nn
->nn_sc_valid
, valid
,
382 nn
->nn_persistent_error
, err
);
385 nn
->nn_sc_valid
= valid
? 1 : 0;
386 nn
->nn_persistent_error
= err
;
388 /* mirrors o2net_tx_can_proceed() */
389 if (nn
->nn_persistent_error
|| nn
->nn_sc_valid
)
390 wake_up(&nn
->nn_sc_wq
);
392 if (!was_err
&& nn
->nn_persistent_error
) {
393 o2quo_conn_err(o2net_num_from_nn(nn
));
394 queue_delayed_work(o2net_wq
, &nn
->nn_still_up
,
395 msecs_to_jiffies(O2NET_QUORUM_DELAY_MS
));
398 if (was_valid
&& !valid
) {
399 printk(KERN_INFO
"o2net: no longer connected to "
400 SC_NODEF_FMT
"\n", SC_NODEF_ARGS(old_sc
));
401 o2net_complete_nodes_nsw(nn
);
404 if (!was_valid
&& valid
) {
405 o2quo_conn_up(o2net_num_from_nn(nn
));
406 /* this is a bit of a hack. we only try reconnecting
407 * when heartbeating starts until we get a connection.
408 * if that connection then dies we don't try reconnecting.
409 * the only way to start connecting again is to down
410 * heartbeat and bring it back up. */
411 cancel_delayed_work(&nn
->nn_connect_expired
);
412 printk(KERN_INFO
"o2net: %s " SC_NODEF_FMT
"\n",
413 o2nm_this_node() > sc
->sc_node
->nd_num
?
414 "connected to" : "accepted connection from",
418 /* trigger the connecting worker func as long as we're not valid,
419 * it will back off if it shouldn't connect. This can be called
420 * from node config teardown and so needs to be careful about
421 * the work queue actually being up. */
422 if (!valid
&& o2net_wq
) {
424 /* delay if we're withing a RECONNECT_DELAY of the
426 delay
= (nn
->nn_last_connect_attempt
+
427 msecs_to_jiffies(O2NET_RECONNECT_DELAY_MS
))
429 if (delay
> msecs_to_jiffies(O2NET_RECONNECT_DELAY_MS
))
431 mlog(ML_CONN
, "queueing conn attempt in %lu jiffies\n", delay
);
432 queue_delayed_work(o2net_wq
, &nn
->nn_connect_work
, delay
);
435 /* keep track of the nn's sc ref for the caller */
436 if ((old_sc
== NULL
) && sc
)
438 if (old_sc
&& (old_sc
!= sc
)) {
439 o2net_sc_queue_work(old_sc
, &old_sc
->sc_shutdown_work
);
444 /* see o2net_register_callbacks() */
445 static void o2net_data_ready(struct sock
*sk
, int bytes
)
447 void (*ready
)(struct sock
*sk
, int bytes
);
449 read_lock(&sk
->sk_callback_lock
);
450 if (sk
->sk_user_data
) {
451 struct o2net_sock_container
*sc
= sk
->sk_user_data
;
452 sclog(sc
, "data_ready hit\n");
453 do_gettimeofday(&sc
->sc_tv_data_ready
);
454 o2net_sc_queue_work(sc
, &sc
->sc_rx_work
);
455 ready
= sc
->sc_data_ready
;
457 ready
= sk
->sk_data_ready
;
459 read_unlock(&sk
->sk_callback_lock
);
464 /* see o2net_register_callbacks() */
465 static void o2net_state_change(struct sock
*sk
)
467 void (*state_change
)(struct sock
*sk
);
468 struct o2net_sock_container
*sc
;
470 read_lock(&sk
->sk_callback_lock
);
471 sc
= sk
->sk_user_data
;
473 state_change
= sk
->sk_state_change
;
477 sclog(sc
, "state_change to %d\n", sk
->sk_state
);
479 state_change
= sc
->sc_state_change
;
481 switch(sk
->sk_state
) {
482 /* ignore connecting sockets as they make progress */
486 case TCP_ESTABLISHED
:
487 o2net_sc_queue_work(sc
, &sc
->sc_connect_work
);
490 o2net_sc_queue_work(sc
, &sc
->sc_shutdown_work
);
494 read_unlock(&sk
->sk_callback_lock
);
499 * we register callbacks so we can queue work on events before calling
500 * the original callbacks. our callbacks our careful to test user_data
501 * to discover when they've reaced with o2net_unregister_callbacks().
503 static void o2net_register_callbacks(struct sock
*sk
,
504 struct o2net_sock_container
*sc
)
506 write_lock_bh(&sk
->sk_callback_lock
);
508 /* accepted sockets inherit the old listen socket data ready */
509 if (sk
->sk_data_ready
== o2net_listen_data_ready
) {
510 sk
->sk_data_ready
= sk
->sk_user_data
;
511 sk
->sk_user_data
= NULL
;
514 BUG_ON(sk
->sk_user_data
!= NULL
);
515 sk
->sk_user_data
= sc
;
518 sc
->sc_data_ready
= sk
->sk_data_ready
;
519 sc
->sc_state_change
= sk
->sk_state_change
;
520 sk
->sk_data_ready
= o2net_data_ready
;
521 sk
->sk_state_change
= o2net_state_change
;
523 write_unlock_bh(&sk
->sk_callback_lock
);
526 static int o2net_unregister_callbacks(struct sock
*sk
,
527 struct o2net_sock_container
*sc
)
531 write_lock_bh(&sk
->sk_callback_lock
);
532 if (sk
->sk_user_data
== sc
) {
534 sk
->sk_user_data
= NULL
;
535 sk
->sk_data_ready
= sc
->sc_data_ready
;
536 sk
->sk_state_change
= sc
->sc_state_change
;
538 write_unlock_bh(&sk
->sk_callback_lock
);
544 * this is a little helper that is called by callers who have seen a problem
545 * with an sc and want to detach it from the nn if someone already hasn't beat
546 * them to it. if an error is given then the shutdown will be persistent
547 * and pending transmits will be canceled.
549 static void o2net_ensure_shutdown(struct o2net_node
*nn
,
550 struct o2net_sock_container
*sc
,
553 spin_lock(&nn
->nn_lock
);
555 o2net_set_nn_state(nn
, NULL
, 0, err
);
556 spin_unlock(&nn
->nn_lock
);
560 * This work queue function performs the blocking parts of socket shutdown. A
561 * few paths lead here. set_nn_state will trigger this callback if it sees an
562 * sc detached from the nn. state_change will also trigger this callback
563 * directly when it sees errors. In that case we need to call set_nn_state
564 * ourselves as state_change couldn't get the nn_lock and call set_nn_state
567 static void o2net_shutdown_sc(struct work_struct
*work
)
569 struct o2net_sock_container
*sc
=
570 container_of(work
, struct o2net_sock_container
,
572 struct o2net_node
*nn
= o2net_nn_from_num(sc
->sc_node
->nd_num
);
574 sclog(sc
, "shutting down\n");
576 /* drop the callbacks ref and call shutdown only once */
577 if (o2net_unregister_callbacks(sc
->sc_sock
->sk
, sc
)) {
578 /* we shouldn't flush as we're in the thread, the
579 * races with pending sc work structs are harmless */
580 del_timer_sync(&sc
->sc_idle_timeout
);
581 o2net_sc_cancel_delayed_work(sc
, &sc
->sc_keepalive_work
);
583 sc
->sc_sock
->ops
->shutdown(sc
->sc_sock
,
584 RCV_SHUTDOWN
|SEND_SHUTDOWN
);
587 /* not fatal so failed connects before the other guy has our
588 * heartbeat can be retried */
589 o2net_ensure_shutdown(nn
, sc
, 0);
593 /* ------------------------------------------------------------ */
595 static int o2net_handler_cmp(struct o2net_msg_handler
*nmh
, u32 msg_type
,
598 int ret
= memcmp(&nmh
->nh_key
, &key
, sizeof(key
));
601 ret
= memcmp(&nmh
->nh_msg_type
, &msg_type
, sizeof(msg_type
));
606 static struct o2net_msg_handler
*
607 o2net_handler_tree_lookup(u32 msg_type
, u32 key
, struct rb_node
***ret_p
,
608 struct rb_node
**ret_parent
)
610 struct rb_node
**p
= &o2net_handler_tree
.rb_node
;
611 struct rb_node
*parent
= NULL
;
612 struct o2net_msg_handler
*nmh
, *ret
= NULL
;
617 nmh
= rb_entry(parent
, struct o2net_msg_handler
, nh_node
);
618 cmp
= o2net_handler_cmp(nmh
, msg_type
, key
);
632 if (ret_parent
!= NULL
)
633 *ret_parent
= parent
;
638 static void o2net_handler_kref_release(struct kref
*kref
)
640 struct o2net_msg_handler
*nmh
;
641 nmh
= container_of(kref
, struct o2net_msg_handler
, nh_kref
);
646 static void o2net_handler_put(struct o2net_msg_handler
*nmh
)
648 kref_put(&nmh
->nh_kref
, o2net_handler_kref_release
);
651 /* max_len is protection for the handler func. incoming messages won't
652 * be given to the handler if their payload is longer than the max. */
653 int o2net_register_handler(u32 msg_type
, u32 key
, u32 max_len
,
654 o2net_msg_handler_func
*func
, void *data
,
655 struct list_head
*unreg_list
)
657 struct o2net_msg_handler
*nmh
= NULL
;
658 struct rb_node
**p
, *parent
;
661 if (max_len
> O2NET_MAX_PAYLOAD_BYTES
) {
662 mlog(0, "max_len for message handler out of range: %u\n",
669 mlog(0, "no message type provided: %u, %p\n", msg_type
, func
);
675 mlog(0, "no message handler provided: %u, %p\n",
681 nmh
= kcalloc(1, sizeof(struct o2net_msg_handler
), GFP_NOFS
);
688 nmh
->nh_func_data
= data
;
689 nmh
->nh_msg_type
= msg_type
;
690 nmh
->nh_max_len
= max_len
;
692 /* the tree and list get this ref.. they're both removed in
693 * unregister when this ref is dropped */
694 kref_init(&nmh
->nh_kref
);
695 INIT_LIST_HEAD(&nmh
->nh_unregister_item
);
697 write_lock(&o2net_handler_lock
);
698 if (o2net_handler_tree_lookup(msg_type
, key
, &p
, &parent
))
701 rb_link_node(&nmh
->nh_node
, parent
, p
);
702 rb_insert_color(&nmh
->nh_node
, &o2net_handler_tree
);
703 list_add_tail(&nmh
->nh_unregister_item
, unreg_list
);
705 mlog(ML_TCP
, "registered handler func %p type %u key %08x\n",
706 func
, msg_type
, key
);
707 /* we've had some trouble with handlers seemingly vanishing. */
708 mlog_bug_on_msg(o2net_handler_tree_lookup(msg_type
, key
, &p
,
710 "couldn't find handler we *just* registerd "
711 "for type %u key %08x\n", msg_type
, key
);
713 write_unlock(&o2net_handler_lock
);
723 EXPORT_SYMBOL_GPL(o2net_register_handler
);
725 void o2net_unregister_handler_list(struct list_head
*list
)
727 struct list_head
*pos
, *n
;
728 struct o2net_msg_handler
*nmh
;
730 write_lock(&o2net_handler_lock
);
731 list_for_each_safe(pos
, n
, list
) {
732 nmh
= list_entry(pos
, struct o2net_msg_handler
,
734 mlog(ML_TCP
, "unregistering handler func %p type %u key %08x\n",
735 nmh
->nh_func
, nmh
->nh_msg_type
, nmh
->nh_key
);
736 rb_erase(&nmh
->nh_node
, &o2net_handler_tree
);
737 list_del_init(&nmh
->nh_unregister_item
);
738 kref_put(&nmh
->nh_kref
, o2net_handler_kref_release
);
740 write_unlock(&o2net_handler_lock
);
742 EXPORT_SYMBOL_GPL(o2net_unregister_handler_list
);
744 static struct o2net_msg_handler
*o2net_handler_get(u32 msg_type
, u32 key
)
746 struct o2net_msg_handler
*nmh
;
748 read_lock(&o2net_handler_lock
);
749 nmh
= o2net_handler_tree_lookup(msg_type
, key
, NULL
, NULL
);
751 kref_get(&nmh
->nh_kref
);
752 read_unlock(&o2net_handler_lock
);
757 /* ------------------------------------------------------------ */
759 static int o2net_recv_tcp_msg(struct socket
*sock
, void *data
, size_t len
)
767 struct msghdr msg
= {
769 .msg_iov
= (struct iovec
*)&vec
,
770 .msg_flags
= MSG_DONTWAIT
,
775 ret
= sock_recvmsg(sock
, &msg
, len
, msg
.msg_flags
);
781 static int o2net_send_tcp_msg(struct socket
*sock
, struct kvec
*vec
,
782 size_t veclen
, size_t total
)
786 struct msghdr msg
= {
787 .msg_iov
= (struct iovec
*)vec
,
788 .msg_iovlen
= veclen
,
798 ret
= sock_sendmsg(sock
, &msg
, total
);
801 mlog(ML_ERROR
, "sendmsg returned %d instead of %zu\n", ret
,
804 ret
= -EPIPE
; /* should be smarter, I bet */
811 mlog(0, "returning error: %d\n", ret
);
815 static void o2net_sendpage(struct o2net_sock_container
*sc
,
816 void *kmalloced_virt
,
819 struct o2net_node
*nn
= o2net_nn_from_num(sc
->sc_node
->nd_num
);
823 ret
= sc
->sc_sock
->ops
->sendpage(sc
->sc_sock
,
824 virt_to_page(kmalloced_virt
),
825 (long)kmalloced_virt
& ~PAGE_MASK
,
828 mlog(ML_ERROR
, "sendpage of size %zu to " SC_NODEF_FMT
829 " failed with %zd\n", size
, SC_NODEF_ARGS(sc
), ret
);
830 o2net_ensure_shutdown(nn
, sc
, 0);
834 static void o2net_init_msg(struct o2net_msg
*msg
, u16 data_len
, u16 msg_type
, u32 key
)
836 memset(msg
, 0, sizeof(struct o2net_msg
));
837 msg
->magic
= cpu_to_be16(O2NET_MSG_MAGIC
);
838 msg
->data_len
= cpu_to_be16(data_len
);
839 msg
->msg_type
= cpu_to_be16(msg_type
);
840 msg
->sys_status
= cpu_to_be32(O2NET_ERR_NONE
);
842 msg
->key
= cpu_to_be32(key
);
845 static int o2net_tx_can_proceed(struct o2net_node
*nn
,
846 struct o2net_sock_container
**sc_ret
,
851 spin_lock(&nn
->nn_lock
);
852 if (nn
->nn_persistent_error
) {
855 *error
= nn
->nn_persistent_error
;
856 } else if (nn
->nn_sc_valid
) {
857 kref_get(&nn
->nn_sc
->sc_kref
);
863 spin_unlock(&nn
->nn_lock
);
868 int o2net_send_message_vec(u32 msg_type
, u32 key
, struct kvec
*caller_vec
,
869 size_t caller_veclen
, u8 target_node
, int *status
)
872 struct o2net_msg
*msg
= NULL
;
873 size_t veclen
, caller_bytes
= 0;
874 struct kvec
*vec
= NULL
;
875 struct o2net_sock_container
*sc
= NULL
;
876 struct o2net_node
*nn
= o2net_nn_from_num(target_node
);
877 struct o2net_status_wait nsw
= {
878 .ns_node_item
= LIST_HEAD_INIT(nsw
.ns_node_item
),
881 if (o2net_wq
== NULL
) {
882 mlog(0, "attempt to tx without o2netd running\n");
887 if (caller_veclen
== 0) {
888 mlog(0, "bad kvec array length\n");
893 caller_bytes
= iov_length((struct iovec
*)caller_vec
, caller_veclen
);
894 if (caller_bytes
> O2NET_MAX_PAYLOAD_BYTES
) {
895 mlog(0, "total payload len %zu too large\n", caller_bytes
);
900 if (target_node
== o2nm_this_node()) {
905 ret
= wait_event_interruptible(nn
->nn_sc_wq
,
906 o2net_tx_can_proceed(nn
, &sc
, &error
));
912 veclen
= caller_veclen
+ 1;
913 vec
= kmalloc(sizeof(struct kvec
) * veclen
, GFP_ATOMIC
);
915 mlog(0, "failed to %zu element kvec!\n", veclen
);
920 msg
= kmalloc(sizeof(struct o2net_msg
), GFP_ATOMIC
);
922 mlog(0, "failed to allocate a o2net_msg!\n");
927 o2net_init_msg(msg
, caller_bytes
, msg_type
, key
);
929 vec
[0].iov_len
= sizeof(struct o2net_msg
);
930 vec
[0].iov_base
= msg
;
931 memcpy(&vec
[1], caller_vec
, caller_veclen
* sizeof(struct kvec
));
933 ret
= o2net_prep_nsw(nn
, &nsw
);
937 msg
->msg_num
= cpu_to_be32(nsw
.ns_id
);
939 /* finally, convert the message header to network byte-order
941 ret
= o2net_send_tcp_msg(sc
->sc_sock
, vec
, veclen
,
942 sizeof(struct o2net_msg
) + caller_bytes
);
943 msglog(msg
, "sending returned %d\n", ret
);
945 mlog(0, "error returned from o2net_send_tcp_msg=%d\n", ret
);
949 /* wait on other node's handler */
950 wait_event(nsw
.ns_wq
, o2net_nsw_completed(nn
, &nsw
));
952 /* Note that we avoid overwriting the callers status return
953 * variable if a system error was reported on the other
954 * side. Callers beware. */
955 ret
= o2net_sys_err_to_errno(nsw
.ns_sys_status
);
957 *status
= nsw
.ns_status
;
959 mlog(0, "woken, returning system status %d, user status %d\n",
968 o2net_complete_nsw(nn
, &nsw
, 0, 0, 0);
971 EXPORT_SYMBOL_GPL(o2net_send_message_vec
);
973 int o2net_send_message(u32 msg_type
, u32 key
, void *data
, u32 len
,
974 u8 target_node
, int *status
)
980 return o2net_send_message_vec(msg_type
, key
, &vec
, 1,
981 target_node
, status
);
983 EXPORT_SYMBOL_GPL(o2net_send_message
);
985 static int o2net_send_status_magic(struct socket
*sock
, struct o2net_msg
*hdr
,
986 enum o2net_system_error syserr
, int err
)
990 .iov_len
= sizeof(struct o2net_msg
),
993 BUG_ON(syserr
>= O2NET_ERR_MAX
);
995 /* leave other fields intact from the incoming message, msg_num
997 hdr
->sys_status
= cpu_to_be32(syserr
);
998 hdr
->status
= cpu_to_be32(err
);
999 hdr
->magic
= cpu_to_be16(O2NET_MSG_STATUS_MAGIC
); // twiddle the magic
1002 msglog(hdr
, "about to send status magic %d\n", err
);
1003 /* hdr has been in host byteorder this whole time */
1004 return o2net_send_tcp_msg(sock
, &vec
, 1, sizeof(struct o2net_msg
));
1007 /* this returns -errno if the header was unknown or too large, etc.
1008 * after this is called the buffer us reused for the next message */
1009 static int o2net_process_message(struct o2net_sock_container
*sc
,
1010 struct o2net_msg
*hdr
)
1012 struct o2net_node
*nn
= o2net_nn_from_num(sc
->sc_node
->nd_num
);
1013 int ret
= 0, handler_status
;
1014 enum o2net_system_error syserr
;
1015 struct o2net_msg_handler
*nmh
= NULL
;
1017 msglog(hdr
, "processing message\n");
1019 o2net_sc_postpone_idle(sc
);
1021 switch(be16_to_cpu(hdr
->magic
)) {
1022 case O2NET_MSG_STATUS_MAGIC
:
1023 /* special type for returning message status */
1024 o2net_complete_nsw(nn
, NULL
,
1025 be32_to_cpu(hdr
->msg_num
),
1026 be32_to_cpu(hdr
->sys_status
),
1027 be32_to_cpu(hdr
->status
));
1029 case O2NET_MSG_KEEP_REQ_MAGIC
:
1030 o2net_sendpage(sc
, o2net_keep_resp
,
1031 sizeof(*o2net_keep_resp
));
1033 case O2NET_MSG_KEEP_RESP_MAGIC
:
1035 case O2NET_MSG_MAGIC
:
1038 msglog(hdr
, "bad magic\n");
1044 /* find a handler for it */
1046 nmh
= o2net_handler_get(be16_to_cpu(hdr
->msg_type
),
1047 be32_to_cpu(hdr
->key
));
1049 mlog(ML_TCP
, "couldn't find handler for type %u key %08x\n",
1050 be16_to_cpu(hdr
->msg_type
), be32_to_cpu(hdr
->key
));
1051 syserr
= O2NET_ERR_NO_HNDLR
;
1055 syserr
= O2NET_ERR_NONE
;
1057 if (be16_to_cpu(hdr
->data_len
) > nmh
->nh_max_len
)
1058 syserr
= O2NET_ERR_OVERFLOW
;
1060 if (syserr
!= O2NET_ERR_NONE
)
1063 do_gettimeofday(&sc
->sc_tv_func_start
);
1064 sc
->sc_msg_key
= be32_to_cpu(hdr
->key
);
1065 sc
->sc_msg_type
= be16_to_cpu(hdr
->msg_type
);
1066 handler_status
= (nmh
->nh_func
)(hdr
, sizeof(struct o2net_msg
) +
1067 be16_to_cpu(hdr
->data_len
),
1069 do_gettimeofday(&sc
->sc_tv_func_stop
);
1072 /* this destroys the hdr, so don't use it after this */
1073 ret
= o2net_send_status_magic(sc
->sc_sock
, hdr
, syserr
,
1076 mlog(0, "sending handler status %d, syserr %d returned %d\n",
1077 handler_status
, syserr
, ret
);
1081 o2net_handler_put(nmh
);
1085 static int o2net_check_handshake(struct o2net_sock_container
*sc
)
1087 struct o2net_handshake
*hand
= page_address(sc
->sc_page
);
1088 struct o2net_node
*nn
= o2net_nn_from_num(sc
->sc_node
->nd_num
);
1090 if (hand
->protocol_version
!= cpu_to_be64(O2NET_PROTOCOL_VERSION
)) {
1091 mlog(ML_NOTICE
, SC_NODEF_FMT
" advertised net protocol "
1092 "version %llu but %llu is required, disconnecting\n",
1094 (unsigned long long)be64_to_cpu(hand
->protocol_version
),
1095 O2NET_PROTOCOL_VERSION
);
1097 /* don't bother reconnecting if its the wrong version. */
1098 o2net_ensure_shutdown(nn
, sc
, -ENOTCONN
);
1102 sc
->sc_handshake_ok
= 1;
1104 spin_lock(&nn
->nn_lock
);
1105 /* set valid and queue the idle timers only if it hasn't been
1106 * shut down already */
1107 if (nn
->nn_sc
== sc
) {
1108 o2net_sc_postpone_idle(sc
);
1109 o2net_set_nn_state(nn
, sc
, 1, 0);
1111 spin_unlock(&nn
->nn_lock
);
1113 /* shift everything up as though it wasn't there */
1114 sc
->sc_page_off
-= sizeof(struct o2net_handshake
);
1115 if (sc
->sc_page_off
)
1116 memmove(hand
, hand
+ 1, sc
->sc_page_off
);
1121 /* this demuxes the queued rx bytes into header or payload bits and calls
1122 * handlers as each full message is read off the socket. it returns -error,
1123 * == 0 eof, or > 0 for progress made.*/
1124 static int o2net_advance_rx(struct o2net_sock_container
*sc
)
1126 struct o2net_msg
*hdr
;
1131 sclog(sc
, "receiving\n");
1132 do_gettimeofday(&sc
->sc_tv_advance_start
);
1134 /* do we need more header? */
1135 if (sc
->sc_page_off
< sizeof(struct o2net_msg
)) {
1136 data
= page_address(sc
->sc_page
) + sc
->sc_page_off
;
1137 datalen
= sizeof(struct o2net_msg
) - sc
->sc_page_off
;
1138 ret
= o2net_recv_tcp_msg(sc
->sc_sock
, data
, datalen
);
1140 sc
->sc_page_off
+= ret
;
1142 /* this working relies on the handshake being
1143 * smaller than the normal message header */
1144 if (sc
->sc_page_off
>= sizeof(struct o2net_handshake
)&&
1145 !sc
->sc_handshake_ok
&& o2net_check_handshake(sc
)) {
1150 /* only swab incoming here.. we can
1151 * only get here once as we cross from
1152 * being under to over */
1153 if (sc
->sc_page_off
== sizeof(struct o2net_msg
)) {
1154 hdr
= page_address(sc
->sc_page
);
1155 if (be16_to_cpu(hdr
->data_len
) >
1156 O2NET_MAX_PAYLOAD_BYTES
)
1164 if (sc
->sc_page_off
< sizeof(struct o2net_msg
)) {
1165 /* oof, still don't have a header */
1169 /* this was swabbed above when we first read it */
1170 hdr
= page_address(sc
->sc_page
);
1172 msglog(hdr
, "at page_off %zu\n", sc
->sc_page_off
);
1174 /* do we need more payload? */
1175 if (sc
->sc_page_off
- sizeof(struct o2net_msg
) < be16_to_cpu(hdr
->data_len
)) {
1176 /* need more payload */
1177 data
= page_address(sc
->sc_page
) + sc
->sc_page_off
;
1178 datalen
= (sizeof(struct o2net_msg
) + be16_to_cpu(hdr
->data_len
)) -
1180 ret
= o2net_recv_tcp_msg(sc
->sc_sock
, data
, datalen
);
1182 sc
->sc_page_off
+= ret
;
1187 if (sc
->sc_page_off
- sizeof(struct o2net_msg
) == be16_to_cpu(hdr
->data_len
)) {
1188 /* we can only get here once, the first time we read
1189 * the payload.. so set ret to progress if the handler
1190 * works out. after calling this the message is toast */
1191 ret
= o2net_process_message(sc
, hdr
);
1194 sc
->sc_page_off
= 0;
1198 sclog(sc
, "ret = %d\n", ret
);
1199 do_gettimeofday(&sc
->sc_tv_advance_stop
);
1203 /* this work func is triggerd by data ready. it reads until it can read no
1204 * more. it interprets 0, eof, as fatal. if data_ready hits while we're doing
1205 * our work the work struct will be marked and we'll be called again. */
1206 static void o2net_rx_until_empty(struct work_struct
*work
)
1208 struct o2net_sock_container
*sc
=
1209 container_of(work
, struct o2net_sock_container
, sc_rx_work
);
1213 ret
= o2net_advance_rx(sc
);
1216 if (ret
<= 0 && ret
!= -EAGAIN
) {
1217 struct o2net_node
*nn
= o2net_nn_from_num(sc
->sc_node
->nd_num
);
1218 sclog(sc
, "saw error %d, closing\n", ret
);
1219 /* not permanent so read failed handshake can retry */
1220 o2net_ensure_shutdown(nn
, sc
, 0);
1226 static int o2net_set_nodelay(struct socket
*sock
)
1235 * Dear unsuspecting programmer,
1237 * Don't use sock_setsockopt() for SOL_TCP. It doesn't check its level
1238 * argument and assumes SOL_SOCKET so, say, your TCP_NODELAY will
1239 * silently turn into SO_DEBUG.
1242 * Keeper of hilariously fragile interfaces.
1244 ret
= sock
->ops
->setsockopt(sock
, SOL_TCP
, TCP_NODELAY
,
1245 (char __user
*)&val
, sizeof(val
));
1251 /* ------------------------------------------------------------ */
1253 /* called when a connect completes and after a sock is accepted. the
1254 * rx path will see the response and mark the sc valid */
1255 static void o2net_sc_connect_completed(struct work_struct
*work
)
1257 struct o2net_sock_container
*sc
=
1258 container_of(work
, struct o2net_sock_container
,
1261 mlog(ML_MSG
, "sc sending handshake with ver %llu id %llx\n",
1262 (unsigned long long)O2NET_PROTOCOL_VERSION
,
1263 (unsigned long long)be64_to_cpu(o2net_hand
->connector_id
));
1265 o2net_sendpage(sc
, o2net_hand
, sizeof(*o2net_hand
));
1269 /* this is called as a work_struct func. */
1270 static void o2net_sc_send_keep_req(struct work_struct
*work
)
1272 struct o2net_sock_container
*sc
=
1273 container_of(work
, struct o2net_sock_container
,
1274 sc_keepalive_work
.work
);
1276 o2net_sendpage(sc
, o2net_keep_req
, sizeof(*o2net_keep_req
));
1280 /* socket shutdown does a del_timer_sync against this as it tears down.
1281 * we can't start this timer until we've got to the point in sc buildup
1282 * where shutdown is going to be involved */
1283 static void o2net_idle_timer(unsigned long data
)
1285 struct o2net_sock_container
*sc
= (struct o2net_sock_container
*)data
;
1288 do_gettimeofday(&now
);
1290 printk(KERN_INFO
"o2net: connection to " SC_NODEF_FMT
" has been idle for 10 "
1291 "seconds, shutting it down.\n", SC_NODEF_ARGS(sc
));
1292 mlog(ML_NOTICE
, "here are some times that might help debug the "
1293 "situation: (tmr %ld.%ld now %ld.%ld dr %ld.%ld adv "
1294 "%ld.%ld:%ld.%ld func (%08x:%u) %ld.%ld:%ld.%ld)\n",
1295 sc
->sc_tv_timer
.tv_sec
, (long) sc
->sc_tv_timer
.tv_usec
,
1296 now
.tv_sec
, (long) now
.tv_usec
,
1297 sc
->sc_tv_data_ready
.tv_sec
, (long) sc
->sc_tv_data_ready
.tv_usec
,
1298 sc
->sc_tv_advance_start
.tv_sec
,
1299 (long) sc
->sc_tv_advance_start
.tv_usec
,
1300 sc
->sc_tv_advance_stop
.tv_sec
,
1301 (long) sc
->sc_tv_advance_stop
.tv_usec
,
1302 sc
->sc_msg_key
, sc
->sc_msg_type
,
1303 sc
->sc_tv_func_start
.tv_sec
, (long) sc
->sc_tv_func_start
.tv_usec
,
1304 sc
->sc_tv_func_stop
.tv_sec
, (long) sc
->sc_tv_func_stop
.tv_usec
);
1306 o2net_sc_queue_work(sc
, &sc
->sc_shutdown_work
);
1309 static void o2net_sc_postpone_idle(struct o2net_sock_container
*sc
)
1311 o2net_sc_cancel_delayed_work(sc
, &sc
->sc_keepalive_work
);
1312 o2net_sc_queue_delayed_work(sc
, &sc
->sc_keepalive_work
,
1313 O2NET_KEEPALIVE_DELAY_SECS
* HZ
);
1314 do_gettimeofday(&sc
->sc_tv_timer
);
1315 mod_timer(&sc
->sc_idle_timeout
,
1316 jiffies
+ (O2NET_IDLE_TIMEOUT_SECS
* HZ
));
1319 /* this work func is kicked whenever a path sets the nn state which doesn't
1320 * have valid set. This includes seeing hb come up, losing a connection,
1321 * having a connect attempt fail, etc. This centralizes the logic which decides
1322 * if a connect attempt should be made or if we should give up and all future
1323 * transmit attempts should fail */
1324 static void o2net_start_connect(struct work_struct
*work
)
1326 struct o2net_node
*nn
=
1327 container_of(work
, struct o2net_node
, nn_connect_work
.work
);
1328 struct o2net_sock_container
*sc
= NULL
;
1329 struct o2nm_node
*node
= NULL
, *mynode
= NULL
;
1330 struct socket
*sock
= NULL
;
1331 struct sockaddr_in myaddr
= {0, }, remoteaddr
= {0, };
1334 /* if we're greater we initiate tx, otherwise we accept */
1335 if (o2nm_this_node() <= o2net_num_from_nn(nn
))
1338 /* watch for racing with tearing a node down */
1339 node
= o2nm_get_node_by_num(o2net_num_from_nn(nn
));
1345 mynode
= o2nm_get_node_by_num(o2nm_this_node());
1346 if (mynode
== NULL
) {
1351 spin_lock(&nn
->nn_lock
);
1352 /* see if we already have one pending or have given up */
1353 stop
= (nn
->nn_sc
|| nn
->nn_persistent_error
);
1354 spin_unlock(&nn
->nn_lock
);
1358 nn
->nn_last_connect_attempt
= jiffies
;
1360 sc
= sc_alloc(node
);
1362 mlog(0, "couldn't allocate sc\n");
1367 ret
= sock_create(PF_INET
, SOCK_STREAM
, IPPROTO_TCP
, &sock
);
1369 mlog(0, "can't create socket: %d\n", ret
);
1372 sc
->sc_sock
= sock
; /* freed by sc_kref_release */
1374 sock
->sk
->sk_allocation
= GFP_ATOMIC
;
1376 myaddr
.sin_family
= AF_INET
;
1377 myaddr
.sin_addr
.s_addr
= (__force u32
)mynode
->nd_ipv4_address
;
1378 myaddr
.sin_port
= (__force u16
)htons(0); /* any port */
1380 ret
= sock
->ops
->bind(sock
, (struct sockaddr
*)&myaddr
,
1383 mlog(ML_ERROR
, "bind failed with %d at address %u.%u.%u.%u\n",
1384 ret
, NIPQUAD(mynode
->nd_ipv4_address
));
1388 ret
= o2net_set_nodelay(sc
->sc_sock
);
1390 mlog(ML_ERROR
, "setting TCP_NODELAY failed with %d\n", ret
);
1394 o2net_register_callbacks(sc
->sc_sock
->sk
, sc
);
1396 spin_lock(&nn
->nn_lock
);
1397 /* handshake completion will set nn->nn_sc_valid */
1398 o2net_set_nn_state(nn
, sc
, 0, 0);
1399 spin_unlock(&nn
->nn_lock
);
1401 remoteaddr
.sin_family
= AF_INET
;
1402 remoteaddr
.sin_addr
.s_addr
= (__force u32
)node
->nd_ipv4_address
;
1403 remoteaddr
.sin_port
= (__force u16
)node
->nd_ipv4_port
;
1405 ret
= sc
->sc_sock
->ops
->connect(sc
->sc_sock
,
1406 (struct sockaddr
*)&remoteaddr
,
1409 if (ret
== -EINPROGRESS
)
1414 mlog(ML_NOTICE
, "connect attempt to " SC_NODEF_FMT
" failed "
1415 "with errno %d\n", SC_NODEF_ARGS(sc
), ret
);
1416 /* 0 err so that another will be queued and attempted
1417 * from set_nn_state */
1419 o2net_ensure_shutdown(nn
, sc
, 0);
1424 o2nm_node_put(node
);
1426 o2nm_node_put(mynode
);
1431 static void o2net_connect_expired(struct work_struct
*work
)
1433 struct o2net_node
*nn
=
1434 container_of(work
, struct o2net_node
, nn_connect_expired
.work
);
1436 spin_lock(&nn
->nn_lock
);
1437 if (!nn
->nn_sc_valid
) {
1438 mlog(ML_ERROR
, "no connection established with node %u after "
1439 "%u seconds, giving up and returning errors.\n",
1440 o2net_num_from_nn(nn
), O2NET_IDLE_TIMEOUT_SECS
);
1442 o2net_set_nn_state(nn
, NULL
, 0, -ENOTCONN
);
1444 spin_unlock(&nn
->nn_lock
);
1447 static void o2net_still_up(struct work_struct
*work
)
1449 struct o2net_node
*nn
=
1450 container_of(work
, struct o2net_node
, nn_still_up
.work
);
1452 o2quo_hb_still_up(o2net_num_from_nn(nn
));
1455 /* ------------------------------------------------------------ */
1457 void o2net_disconnect_node(struct o2nm_node
*node
)
1459 struct o2net_node
*nn
= o2net_nn_from_num(node
->nd_num
);
1461 /* don't reconnect until it's heartbeating again */
1462 spin_lock(&nn
->nn_lock
);
1463 o2net_set_nn_state(nn
, NULL
, 0, -ENOTCONN
);
1464 spin_unlock(&nn
->nn_lock
);
1467 cancel_delayed_work(&nn
->nn_connect_expired
);
1468 cancel_delayed_work(&nn
->nn_connect_work
);
1469 cancel_delayed_work(&nn
->nn_still_up
);
1470 flush_workqueue(o2net_wq
);
1474 static void o2net_hb_node_down_cb(struct o2nm_node
*node
, int node_num
,
1477 o2quo_hb_down(node_num
);
1479 if (node_num
!= o2nm_this_node())
1480 o2net_disconnect_node(node
);
1483 static void o2net_hb_node_up_cb(struct o2nm_node
*node
, int node_num
,
1486 struct o2net_node
*nn
= o2net_nn_from_num(node_num
);
1488 o2quo_hb_up(node_num
);
1490 /* ensure an immediate connect attempt */
1491 nn
->nn_last_connect_attempt
= jiffies
-
1492 (msecs_to_jiffies(O2NET_RECONNECT_DELAY_MS
) + 1);
1494 if (node_num
!= o2nm_this_node()) {
1495 /* heartbeat doesn't work unless a local node number is
1496 * configured and doing so brings up the o2net_wq, so we can
1498 queue_delayed_work(o2net_wq
, &nn
->nn_connect_expired
,
1499 O2NET_IDLE_TIMEOUT_SECS
* HZ
);
1501 /* believe it or not, accept and node hearbeating testing
1502 * can succeed for this node before we got here.. so
1503 * only use set_nn_state to clear the persistent error
1504 * if that hasn't already happened */
1505 spin_lock(&nn
->nn_lock
);
1506 if (nn
->nn_persistent_error
)
1507 o2net_set_nn_state(nn
, NULL
, 0, 0);
1508 spin_unlock(&nn
->nn_lock
);
1512 void o2net_unregister_hb_callbacks(void)
1516 ret
= o2hb_unregister_callback(&o2net_hb_up
);
1518 mlog(ML_ERROR
, "Status return %d unregistering heartbeat up "
1519 "callback!\n", ret
);
1521 ret
= o2hb_unregister_callback(&o2net_hb_down
);
1523 mlog(ML_ERROR
, "Status return %d unregistering heartbeat down "
1524 "callback!\n", ret
);
1527 int o2net_register_hb_callbacks(void)
1531 o2hb_setup_callback(&o2net_hb_down
, O2HB_NODE_DOWN_CB
,
1532 o2net_hb_node_down_cb
, NULL
, O2NET_HB_PRI
);
1533 o2hb_setup_callback(&o2net_hb_up
, O2HB_NODE_UP_CB
,
1534 o2net_hb_node_up_cb
, NULL
, O2NET_HB_PRI
);
1536 ret
= o2hb_register_callback(&o2net_hb_up
);
1538 ret
= o2hb_register_callback(&o2net_hb_down
);
1541 o2net_unregister_hb_callbacks();
1546 /* ------------------------------------------------------------ */
1548 static int o2net_accept_one(struct socket
*sock
)
1551 struct sockaddr_in sin
;
1552 struct socket
*new_sock
= NULL
;
1553 struct o2nm_node
*node
= NULL
;
1554 struct o2net_sock_container
*sc
= NULL
;
1555 struct o2net_node
*nn
;
1557 BUG_ON(sock
== NULL
);
1558 ret
= sock_create_lite(sock
->sk
->sk_family
, sock
->sk
->sk_type
,
1559 sock
->sk
->sk_protocol
, &new_sock
);
1563 new_sock
->type
= sock
->type
;
1564 new_sock
->ops
= sock
->ops
;
1565 ret
= sock
->ops
->accept(sock
, new_sock
, O_NONBLOCK
);
1569 new_sock
->sk
->sk_allocation
= GFP_ATOMIC
;
1571 ret
= o2net_set_nodelay(new_sock
);
1573 mlog(ML_ERROR
, "setting TCP_NODELAY failed with %d\n", ret
);
1578 ret
= new_sock
->ops
->getname(new_sock
, (struct sockaddr
*) &sin
,
1583 node
= o2nm_get_node_by_ip((__force __be32
)sin
.sin_addr
.s_addr
);
1585 mlog(ML_NOTICE
, "attempt to connect from unknown node at "
1586 "%u.%u.%u.%u:%d\n", NIPQUAD(sin
.sin_addr
.s_addr
),
1587 ntohs((__force __be16
)sin
.sin_port
));
1592 if (o2nm_this_node() > node
->nd_num
) {
1593 mlog(ML_NOTICE
, "unexpected connect attempted from a lower "
1594 "numbered node '%s' at " "%u.%u.%u.%u:%d with num %u\n",
1595 node
->nd_name
, NIPQUAD(sin
.sin_addr
.s_addr
),
1596 ntohs((__force __be16
)sin
.sin_port
), node
->nd_num
);
1601 /* this happens all the time when the other node sees our heartbeat
1602 * and tries to connect before we see their heartbeat */
1603 if (!o2hb_check_node_heartbeating_from_callback(node
->nd_num
)) {
1604 mlog(ML_CONN
, "attempt to connect from node '%s' at "
1605 "%u.%u.%u.%u:%d but it isn't heartbeating\n",
1606 node
->nd_name
, NIPQUAD(sin
.sin_addr
.s_addr
),
1607 ntohs((__force __be16
)sin
.sin_port
));
1612 nn
= o2net_nn_from_num(node
->nd_num
);
1614 spin_lock(&nn
->nn_lock
);
1619 spin_unlock(&nn
->nn_lock
);
1621 mlog(ML_NOTICE
, "attempt to connect from node '%s' at "
1622 "%u.%u.%u.%u:%d but it already has an open connection\n",
1623 node
->nd_name
, NIPQUAD(sin
.sin_addr
.s_addr
),
1624 ntohs((__force __be16
)sin
.sin_port
));
1628 sc
= sc_alloc(node
);
1634 sc
->sc_sock
= new_sock
;
1637 spin_lock(&nn
->nn_lock
);
1638 o2net_set_nn_state(nn
, sc
, 0, 0);
1639 spin_unlock(&nn
->nn_lock
);
1641 o2net_register_callbacks(sc
->sc_sock
->sk
, sc
);
1642 o2net_sc_queue_work(sc
, &sc
->sc_rx_work
);
1644 o2net_sendpage(sc
, o2net_hand
, sizeof(*o2net_hand
));
1648 sock_release(new_sock
);
1650 o2nm_node_put(node
);
1656 static void o2net_accept_many(struct work_struct
*work
)
1658 struct socket
*sock
= o2net_listen_sock
;
1659 while (o2net_accept_one(sock
) == 0)
1663 static void o2net_listen_data_ready(struct sock
*sk
, int bytes
)
1665 void (*ready
)(struct sock
*sk
, int bytes
);
1667 read_lock(&sk
->sk_callback_lock
);
1668 ready
= sk
->sk_user_data
;
1669 if (ready
== NULL
) { /* check for teardown race */
1670 ready
= sk
->sk_data_ready
;
1674 /* ->sk_data_ready is also called for a newly established child socket
1675 * before it has been accepted and the acceptor has set up their
1676 * data_ready.. we only want to queue listen work for our listening
1678 if (sk
->sk_state
== TCP_LISTEN
) {
1679 mlog(ML_TCP
, "bytes: %d\n", bytes
);
1680 queue_work(o2net_wq
, &o2net_listen_work
);
1684 read_unlock(&sk
->sk_callback_lock
);
1688 static int o2net_open_listening_sock(__be16 port
)
1690 struct socket
*sock
= NULL
;
1692 struct sockaddr_in sin
= {
1693 .sin_family
= PF_INET
,
1694 .sin_addr
= { .s_addr
= (__force u32
)htonl(INADDR_ANY
) },
1695 .sin_port
= (__force u16
)port
,
1698 ret
= sock_create(PF_INET
, SOCK_STREAM
, IPPROTO_TCP
, &sock
);
1700 mlog(ML_ERROR
, "unable to create socket, ret=%d\n", ret
);
1704 sock
->sk
->sk_allocation
= GFP_ATOMIC
;
1706 write_lock_bh(&sock
->sk
->sk_callback_lock
);
1707 sock
->sk
->sk_user_data
= sock
->sk
->sk_data_ready
;
1708 sock
->sk
->sk_data_ready
= o2net_listen_data_ready
;
1709 write_unlock_bh(&sock
->sk
->sk_callback_lock
);
1711 o2net_listen_sock
= sock
;
1712 INIT_WORK(&o2net_listen_work
, o2net_accept_many
);
1714 sock
->sk
->sk_reuse
= 1;
1715 ret
= sock
->ops
->bind(sock
, (struct sockaddr
*)&sin
, sizeof(sin
));
1717 mlog(ML_ERROR
, "unable to bind socket to port %d, ret=%d\n",
1722 ret
= sock
->ops
->listen(sock
, 64);
1724 mlog(ML_ERROR
, "unable to listen on port %d, ret=%d\n",
1730 o2net_listen_sock
= NULL
;
1738 * called from node manager when we should bring up our network listening
1739 * socket. node manager handles all the serialization to only call this
1740 * once and to match it with o2net_stop_listening(). note,
1741 * o2nm_this_node() doesn't work yet as we're being called while it
1744 int o2net_start_listening(struct o2nm_node
*node
)
1748 BUG_ON(o2net_wq
!= NULL
);
1749 BUG_ON(o2net_listen_sock
!= NULL
);
1751 mlog(ML_KTHREAD
, "starting o2net thread...\n");
1752 o2net_wq
= create_singlethread_workqueue("o2net");
1753 if (o2net_wq
== NULL
) {
1754 mlog(ML_ERROR
, "unable to launch o2net thread\n");
1755 return -ENOMEM
; /* ? */
1758 ret
= o2net_open_listening_sock(node
->nd_ipv4_port
);
1760 destroy_workqueue(o2net_wq
);
1763 o2quo_conn_up(node
->nd_num
);
1768 /* again, o2nm_this_node() doesn't work here as we're involved in
1769 * tearing it down */
1770 void o2net_stop_listening(struct o2nm_node
*node
)
1772 struct socket
*sock
= o2net_listen_sock
;
1775 BUG_ON(o2net_wq
== NULL
);
1776 BUG_ON(o2net_listen_sock
== NULL
);
1778 /* stop the listening socket from generating work */
1779 write_lock_bh(&sock
->sk
->sk_callback_lock
);
1780 sock
->sk
->sk_data_ready
= sock
->sk
->sk_user_data
;
1781 sock
->sk
->sk_user_data
= NULL
;
1782 write_unlock_bh(&sock
->sk
->sk_callback_lock
);
1784 for (i
= 0; i
< ARRAY_SIZE(o2net_nodes
); i
++) {
1785 struct o2nm_node
*node
= o2nm_get_node_by_num(i
);
1787 o2net_disconnect_node(node
);
1788 o2nm_node_put(node
);
1792 /* finish all work and tear down the work queue */
1793 mlog(ML_KTHREAD
, "waiting for o2net thread to exit....\n");
1794 destroy_workqueue(o2net_wq
);
1797 sock_release(o2net_listen_sock
);
1798 o2net_listen_sock
= NULL
;
1800 o2quo_conn_err(node
->nd_num
);
1803 /* ------------------------------------------------------------ */
1805 int o2net_init(void)
1811 o2net_hand
= kcalloc(1, sizeof(struct o2net_handshake
), GFP_KERNEL
);
1812 o2net_keep_req
= kcalloc(1, sizeof(struct o2net_msg
), GFP_KERNEL
);
1813 o2net_keep_resp
= kcalloc(1, sizeof(struct o2net_msg
), GFP_KERNEL
);
1814 if (!o2net_hand
|| !o2net_keep_req
|| !o2net_keep_resp
) {
1816 kfree(o2net_keep_req
);
1817 kfree(o2net_keep_resp
);
1821 o2net_hand
->protocol_version
= cpu_to_be64(O2NET_PROTOCOL_VERSION
);
1822 o2net_hand
->connector_id
= cpu_to_be64(1);
1824 o2net_keep_req
->magic
= cpu_to_be16(O2NET_MSG_KEEP_REQ_MAGIC
);
1825 o2net_keep_resp
->magic
= cpu_to_be16(O2NET_MSG_KEEP_RESP_MAGIC
);
1827 for (i
= 0; i
< ARRAY_SIZE(o2net_nodes
); i
++) {
1828 struct o2net_node
*nn
= o2net_nn_from_num(i
);
1830 spin_lock_init(&nn
->nn_lock
);
1831 INIT_DELAYED_WORK(&nn
->nn_connect_work
, o2net_start_connect
);
1832 INIT_DELAYED_WORK(&nn
->nn_connect_expired
,
1833 o2net_connect_expired
);
1834 INIT_DELAYED_WORK(&nn
->nn_still_up
, o2net_still_up
);
1835 /* until we see hb from a node we'll return einval */
1836 nn
->nn_persistent_error
= -ENOTCONN
;
1837 init_waitqueue_head(&nn
->nn_sc_wq
);
1838 idr_init(&nn
->nn_status_idr
);
1839 INIT_LIST_HEAD(&nn
->nn_status_list
);
1845 void o2net_exit(void)
1849 kfree(o2net_keep_req
);
1850 kfree(o2net_keep_resp
);