2 * net/tipc/port.c: TIPC port code
4 * Copyright (c) 1992-2007, 2014, Ericsson AB
5 * Copyright (c) 2004-2008, 2010-2013, Wind River Systems
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are met:
11 * 1. Redistributions of source code must retain the above copyright
12 * notice, this list of conditions and the following disclaimer.
13 * 2. Redistributions in binary form must reproduce the above copyright
14 * notice, this list of conditions and the following disclaimer in the
15 * documentation and/or other materials provided with the distribution.
16 * 3. Neither the names of the copyright holders nor the names of its
17 * contributors may be used to endorse or promote products derived from
18 * this software without specific prior written permission.
20 * Alternatively, this software may be distributed under the terms of the
21 * GNU General Public License ("GPL") version 2 as published by the Free
22 * Software Foundation.
24 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
25 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
26 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
27 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
28 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
29 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
30 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
31 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
32 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
33 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
34 * POSSIBILITY OF SUCH DAMAGE.
40 #include "name_table.h"
43 /* Connection management: */
44 #define PROBING_INTERVAL 3600000 /* [ms] => 1 h */
48 #define MAX_REJECT_SIZE 1024
50 DEFINE_SPINLOCK(tipc_port_list_lock);
52 static LIST_HEAD(ports);
53 static void port_handle_node_down(unsigned long ref);
54 static struct sk_buff *port_build_self_abort_msg(struct tipc_port *, u32 err);
55 static struct sk_buff *port_build_peer_abort_msg(struct tipc_port *, u32 err);
56 static void port_timeout(unsigned long ref);
59 * tipc_port_peer_msg - verify message was sent by connected port's peer
61 * Handles cases where the node's network address has changed from
62 * the default of <0.0.0> to its configured setting.
64 int tipc_port_peer_msg(struct tipc_port *p_ptr, struct tipc_msg *msg)
69 if (msg_origport(msg) != tipc_port_peerport(p_ptr))
72 orignode = msg_orignode(msg);
73 peernode = tipc_port_peernode(p_ptr);
74 return (orignode == peernode) ||
75 (!orignode && (peernode == tipc_own_addr)) ||
76 (!peernode && (orignode == tipc_own_addr));
80 * tipc_port_mcast_xmit - send a multicast message to local and remote
83 int tipc_port_mcast_xmit(struct tipc_port *oport,
84 struct tipc_name_seq const *seq,
85 struct iovec const *msg_sect,
90 struct sk_buff *ibuf = NULL;
91 struct tipc_port_list dports = {0, NULL, };
95 /* Create multicast message */
97 msg_set_type(hdr, TIPC_MCAST_MSG);
98 msg_set_lookup_scope(hdr, TIPC_CLUSTER_SCOPE);
99 msg_set_destport(hdr, 0);
100 msg_set_destnode(hdr, 0);
101 msg_set_nametype(hdr, seq->type);
102 msg_set_namelower(hdr, seq->lower);
103 msg_set_nameupper(hdr, seq->upper);
104 msg_set_hdr_sz(hdr, MCAST_H_SIZE);
105 res = tipc_msg_build(hdr, msg_sect, len, MAX_MSG_SIZE, &buf);
109 /* Figure out where to send multicast message */
110 ext_targets = tipc_nametbl_mc_translate(seq->type, seq->lower, seq->upper,
111 TIPC_NODE_SCOPE, &dports);
113 /* Send message to destinations (duplicate it only if necessary) */
115 if (dports.count != 0) {
116 ibuf = skb_copy(buf, GFP_ATOMIC);
118 tipc_port_list_free(&dports);
123 res = tipc_bclink_xmit(buf);
124 if ((res < 0) && (dports.count != 0))
132 tipc_port_mcast_rcv(ibuf, &dports);
134 tipc_port_list_free(&dports);
140 * tipc_port_mcast_rcv - deliver multicast message to all destination ports
142 * If there is no port list, perform a lookup to create one
144 void tipc_port_mcast_rcv(struct sk_buff *buf, struct tipc_port_list *dp)
146 struct tipc_msg *msg;
147 struct tipc_port_list dports = {0, NULL, };
148 struct tipc_port_list *item = dp;
153 /* Create destination port list, if one wasn't supplied */
155 tipc_nametbl_mc_translate(msg_nametype(msg),
163 /* Deliver a copy of message to each destination port */
164 if (dp->count != 0) {
165 msg_set_destnode(msg, tipc_own_addr);
166 if (dp->count == 1) {
167 msg_set_destport(msg, dp->ports[0]);
169 tipc_port_list_free(dp);
172 for (; cnt < dp->count; cnt++) {
173 int index = cnt % PLSIZE;
174 struct sk_buff *b = skb_clone(buf, GFP_ATOMIC);
177 pr_warn("Unable to deliver multicast message(s)\n");
180 if ((index == 0) && (cnt != 0))
182 msg_set_destport(buf_msg(b), item->ports[index]);
188 tipc_port_list_free(dp);
192 void tipc_port_wakeup(struct tipc_port *port)
194 tipc_sock_wakeup(tipc_port_to_sock(port));
197 /* tipc_port_init - intiate TIPC port and lock it
199 * Returns obtained reference if initialization is successful, zero otherwise
201 u32 tipc_port_init(struct tipc_port *p_ptr,
202 const unsigned int importance)
204 struct tipc_msg *msg;
207 ref = tipc_ref_acquire(p_ptr, &p_ptr->lock);
209 pr_warn("Port registration failed, ref. table exhausted\n");
213 p_ptr->max_pkt = MAX_PKT_DEFAULT;
216 INIT_LIST_HEAD(&p_ptr->wait_list);
217 INIT_LIST_HEAD(&p_ptr->subscription.nodesub_list);
218 k_init_timer(&p_ptr->timer, (Handler)port_timeout, ref);
219 INIT_LIST_HEAD(&p_ptr->publications);
220 INIT_LIST_HEAD(&p_ptr->port_list);
223 * Must hold port list lock while initializing message header template
224 * to ensure a change to node's own network address doesn't result
225 * in template containing out-dated network address information
227 spin_lock_bh(&tipc_port_list_lock);
229 tipc_msg_init(msg, importance, TIPC_NAMED_MSG, NAMED_H_SIZE, 0);
230 msg_set_origport(msg, ref);
231 list_add_tail(&p_ptr->port_list, &ports);
232 spin_unlock_bh(&tipc_port_list_lock);
236 void tipc_port_destroy(struct tipc_port *p_ptr)
238 struct sk_buff *buf = NULL;
239 struct tipc_msg *msg = NULL;
242 tipc_withdraw(p_ptr, 0, NULL);
244 spin_lock_bh(p_ptr->lock);
245 tipc_ref_discard(p_ptr->ref);
246 spin_unlock_bh(p_ptr->lock);
248 k_cancel_timer(&p_ptr->timer);
249 if (p_ptr->connected) {
250 buf = port_build_peer_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
251 tipc_nodesub_unsubscribe(&p_ptr->subscription);
253 peer = msg_destnode(msg);
254 tipc_link_xmit2(buf, peer, msg_link_selector(msg));
256 spin_lock_bh(&tipc_port_list_lock);
257 list_del(&p_ptr->port_list);
258 list_del(&p_ptr->wait_list);
259 spin_unlock_bh(&tipc_port_list_lock);
260 k_term_timer(&p_ptr->timer);
264 * port_build_proto_msg(): create connection protocol message for port
266 * On entry the port must be locked and connected.
268 static struct sk_buff *port_build_proto_msg(struct tipc_port *p_ptr,
272 struct tipc_msg *msg;
274 buf = tipc_buf_acquire(INT_H_SIZE);
277 tipc_msg_init(msg, CONN_MANAGER, type, INT_H_SIZE,
278 tipc_port_peernode(p_ptr));
279 msg_set_destport(msg, tipc_port_peerport(p_ptr));
280 msg_set_origport(msg, p_ptr->ref);
281 msg_set_msgcnt(msg, ack);
287 static void port_timeout(unsigned long ref)
289 struct tipc_port *p_ptr = tipc_port_lock(ref);
290 struct sk_buff *buf = NULL;
291 struct tipc_msg *msg = NULL;
296 if (!p_ptr->connected) {
297 tipc_port_unlock(p_ptr);
301 /* Last probe answered ? */
302 if (p_ptr->probing_state == PROBING) {
303 buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
305 buf = port_build_proto_msg(p_ptr, CONN_PROBE, 0);
306 p_ptr->probing_state = PROBING;
307 k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
309 tipc_port_unlock(p_ptr);
311 tipc_link_xmit2(buf, msg_destnode(msg), msg_link_selector(msg));
315 static void port_handle_node_down(unsigned long ref)
317 struct tipc_port *p_ptr = tipc_port_lock(ref);
318 struct sk_buff *buf = NULL;
319 struct tipc_msg *msg = NULL;
323 buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_NODE);
324 tipc_port_unlock(p_ptr);
326 tipc_link_xmit2(buf, msg_destnode(msg), msg_link_selector(msg));
330 static struct sk_buff *port_build_self_abort_msg(struct tipc_port *p_ptr, u32 err)
332 struct sk_buff *buf = port_build_peer_abort_msg(p_ptr, err);
335 struct tipc_msg *msg = buf_msg(buf);
336 msg_swap_words(msg, 4, 5);
337 msg_swap_words(msg, 6, 7);
344 static struct sk_buff *port_build_peer_abort_msg(struct tipc_port *p_ptr, u32 err)
347 struct tipc_msg *msg;
350 if (!p_ptr->connected)
353 buf = tipc_buf_acquire(BASIC_H_SIZE);
356 memcpy(msg, &p_ptr->phdr, BASIC_H_SIZE);
357 msg_set_hdr_sz(msg, BASIC_H_SIZE);
358 msg_set_size(msg, BASIC_H_SIZE);
359 imp = msg_importance(msg);
360 if (imp < TIPC_CRITICAL_IMPORTANCE)
361 msg_set_importance(msg, ++imp);
362 msg_set_errcode(msg, err);
368 void tipc_port_proto_rcv(struct tipc_port *p_ptr, struct sk_buff *buf)
370 struct tipc_msg *msg = buf_msg(buf);
371 struct sk_buff *r_buf = NULL;
372 u32 destport = msg_destport(msg);
375 /* Validate connection */
376 if (!p_ptr || !p_ptr->connected || !tipc_port_peer_msg(p_ptr, msg)) {
377 r_buf = tipc_buf_acquire(BASIC_H_SIZE);
379 msg = buf_msg(r_buf);
380 tipc_msg_init(msg, TIPC_HIGH_IMPORTANCE, TIPC_CONN_MSG,
381 BASIC_H_SIZE, msg_orignode(msg));
382 msg_set_errcode(msg, TIPC_ERR_NO_PORT);
383 msg_set_origport(msg, destport);
384 msg_set_destport(msg, msg_origport(msg));
389 /* Process protocol message sent by peer */
390 switch (msg_type(msg)) {
392 wakeable = tipc_port_congested(p_ptr) && p_ptr->congested;
393 p_ptr->acked += msg_msgcnt(msg);
394 if (!tipc_port_congested(p_ptr)) {
395 p_ptr->congested = 0;
397 tipc_port_wakeup(p_ptr);
401 r_buf = port_build_proto_msg(p_ptr, CONN_PROBE_REPLY, 0);
404 /* CONN_PROBE_REPLY or unrecognized - no action required */
407 p_ptr->probing_state = CONFIRMED;
409 tipc_link_xmit2(r_buf, msg_destnode(msg), msg_link_selector(msg));
413 static int port_print(struct tipc_port *p_ptr, char *buf, int len, int full_id)
415 struct publication *publ;
419 ret = tipc_snprintf(buf, len, "<%u.%u.%u:%u>:",
420 tipc_zone(tipc_own_addr),
421 tipc_cluster(tipc_own_addr),
422 tipc_node(tipc_own_addr), p_ptr->ref);
424 ret = tipc_snprintf(buf, len, "%-10u:", p_ptr->ref);
426 if (p_ptr->connected) {
427 u32 dport = tipc_port_peerport(p_ptr);
428 u32 destnode = tipc_port_peernode(p_ptr);
430 ret += tipc_snprintf(buf + ret, len - ret,
431 " connected to <%u.%u.%u:%u>",
433 tipc_cluster(destnode),
434 tipc_node(destnode), dport);
435 if (p_ptr->conn_type != 0)
436 ret += tipc_snprintf(buf + ret, len - ret,
437 " via {%u,%u}", p_ptr->conn_type,
438 p_ptr->conn_instance);
439 } else if (p_ptr->published) {
440 ret += tipc_snprintf(buf + ret, len - ret, " bound to");
441 list_for_each_entry(publ, &p_ptr->publications, pport_list) {
442 if (publ->lower == publ->upper)
443 ret += tipc_snprintf(buf + ret, len - ret,
444 " {%u,%u}", publ->type,
447 ret += tipc_snprintf(buf + ret, len - ret,
448 " {%u,%u,%u}", publ->type,
449 publ->lower, publ->upper);
452 ret += tipc_snprintf(buf + ret, len - ret, "\n");
456 struct sk_buff *tipc_port_get_ports(void)
459 struct tlv_desc *rep_tlv;
462 struct tipc_port *p_ptr;
465 buf = tipc_cfg_reply_alloc(TLV_SPACE(ULTRA_STRING_MAX_LEN));
468 rep_tlv = (struct tlv_desc *)buf->data;
469 pb = TLV_DATA(rep_tlv);
470 pb_len = ULTRA_STRING_MAX_LEN;
472 spin_lock_bh(&tipc_port_list_lock);
473 list_for_each_entry(p_ptr, &ports, port_list) {
474 spin_lock_bh(p_ptr->lock);
475 str_len += port_print(p_ptr, pb, pb_len, 0);
476 spin_unlock_bh(p_ptr->lock);
478 spin_unlock_bh(&tipc_port_list_lock);
479 str_len += 1; /* for "\0" */
480 skb_put(buf, TLV_SPACE(str_len));
481 TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
486 void tipc_port_reinit(void)
488 struct tipc_port *p_ptr;
489 struct tipc_msg *msg;
491 spin_lock_bh(&tipc_port_list_lock);
492 list_for_each_entry(p_ptr, &ports, port_list) {
494 msg_set_prevnode(msg, tipc_own_addr);
495 msg_set_orignode(msg, tipc_own_addr);
497 spin_unlock_bh(&tipc_port_list_lock);
500 void tipc_acknowledge(u32 ref, u32 ack)
502 struct tipc_port *p_ptr;
503 struct sk_buff *buf = NULL;
504 struct tipc_msg *msg;
506 p_ptr = tipc_port_lock(ref);
509 if (p_ptr->connected) {
510 p_ptr->conn_unacked -= ack;
511 buf = port_build_proto_msg(p_ptr, CONN_ACK, ack);
513 tipc_port_unlock(p_ptr);
517 tipc_link_xmit2(buf, msg_destnode(msg), msg_link_selector(msg));
520 int tipc_publish(struct tipc_port *p_ptr, unsigned int scope,
521 struct tipc_name_seq const *seq)
523 struct publication *publ;
526 if (p_ptr->connected)
528 key = p_ptr->ref + p_ptr->pub_count + 1;
529 if (key == p_ptr->ref)
532 publ = tipc_nametbl_publish(seq->type, seq->lower, seq->upper,
533 scope, p_ptr->ref, key);
535 list_add(&publ->pport_list, &p_ptr->publications);
537 p_ptr->published = 1;
543 int tipc_withdraw(struct tipc_port *p_ptr, unsigned int scope,
544 struct tipc_name_seq const *seq)
546 struct publication *publ;
547 struct publication *tpubl;
551 list_for_each_entry_safe(publ, tpubl,
552 &p_ptr->publications, pport_list) {
553 tipc_nametbl_withdraw(publ->type, publ->lower,
554 publ->ref, publ->key);
558 list_for_each_entry_safe(publ, tpubl,
559 &p_ptr->publications, pport_list) {
560 if (publ->scope != scope)
562 if (publ->type != seq->type)
564 if (publ->lower != seq->lower)
566 if (publ->upper != seq->upper)
568 tipc_nametbl_withdraw(publ->type, publ->lower,
569 publ->ref, publ->key);
574 if (list_empty(&p_ptr->publications))
575 p_ptr->published = 0;
579 int tipc_port_connect(u32 ref, struct tipc_portid const *peer)
581 struct tipc_port *p_ptr;
584 p_ptr = tipc_port_lock(ref);
587 res = __tipc_port_connect(ref, p_ptr, peer);
588 tipc_port_unlock(p_ptr);
593 * __tipc_port_connect - connect to a remote peer
595 * Port must be locked.
597 int __tipc_port_connect(u32 ref, struct tipc_port *p_ptr,
598 struct tipc_portid const *peer)
600 struct tipc_msg *msg;
603 if (p_ptr->published || p_ptr->connected)
609 msg_set_destnode(msg, peer->node);
610 msg_set_destport(msg, peer->ref);
611 msg_set_type(msg, TIPC_CONN_MSG);
612 msg_set_lookup_scope(msg, 0);
613 msg_set_hdr_sz(msg, SHORT_H_SIZE);
615 p_ptr->probing_interval = PROBING_INTERVAL;
616 p_ptr->probing_state = CONFIRMED;
617 p_ptr->connected = 1;
618 k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
620 tipc_nodesub_subscribe(&p_ptr->subscription, peer->node,
621 (void *)(unsigned long)ref,
622 (net_ev_handler)port_handle_node_down);
625 p_ptr->max_pkt = tipc_node_get_mtu(peer->node, ref);
630 * __tipc_disconnect - disconnect port from peer
632 * Port must be locked.
634 int __tipc_port_disconnect(struct tipc_port *tp_ptr)
636 if (tp_ptr->connected) {
637 tp_ptr->connected = 0;
638 /* let timer expire on it's own to avoid deadlock! */
639 tipc_nodesub_unsubscribe(&tp_ptr->subscription);
647 * tipc_port_disconnect(): Disconnect port form peer.
648 * This is a node local operation.
650 int tipc_port_disconnect(u32 ref)
652 struct tipc_port *p_ptr;
655 p_ptr = tipc_port_lock(ref);
658 res = __tipc_port_disconnect(p_ptr);
659 tipc_port_unlock(p_ptr);
664 * tipc_port_shutdown(): Send a SHUTDOWN msg to peer and disconnect
666 int tipc_port_shutdown(u32 ref)
668 struct tipc_msg *msg;
669 struct tipc_port *p_ptr;
670 struct sk_buff *buf = NULL;
672 p_ptr = tipc_port_lock(ref);
676 buf = port_build_peer_abort_msg(p_ptr, TIPC_CONN_SHUTDOWN);
677 tipc_port_unlock(p_ptr);
679 tipc_link_xmit2(buf, msg_destnode(msg), msg_link_selector(msg));
680 return tipc_port_disconnect(ref);