Linux Audio

Check our new training course

Loading...
Note: File does not exist in v4.17.
  1/*
  2 * net/tipc/port.c: TIPC port code
  3 *
  4 * Copyright (c) 1992-2007, 2014, Ericsson AB
  5 * Copyright (c) 2004-2008, 2010-2013, Wind River Systems
  6 * All rights reserved.
  7 *
  8 * Redistribution and use in source and binary forms, with or without
  9 * modification, are permitted provided that the following conditions are met:
 10 *
 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.
 19 *
 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.
 23 *
 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.
 35 */
 36
 37#include "core.h"
 38#include "config.h"
 39#include "port.h"
 40#include "name_table.h"
 41#include "socket.h"
 42
 43/* Connection management: */
 44#define PROBING_INTERVAL 3600000	/* [ms] => 1 h */
 45#define CONFIRMED 0
 46#define PROBING 1
 47
 48#define MAX_REJECT_SIZE 1024
 49
 50DEFINE_SPINLOCK(tipc_port_list_lock);
 51
 52static LIST_HEAD(ports);
 53static void port_handle_node_down(unsigned long ref);
 54static struct sk_buff *port_build_self_abort_msg(struct tipc_port *, u32 err);
 55static struct sk_buff *port_build_peer_abort_msg(struct tipc_port *, u32 err);
 56static void port_timeout(unsigned long ref);
 57
 58/**
 59 * tipc_port_peer_msg - verify message was sent by connected port's peer
 60 *
 61 * Handles cases where the node's network address has changed from
 62 * the default of <0.0.0> to its configured setting.
 63 */
 64int tipc_port_peer_msg(struct tipc_port *p_ptr, struct tipc_msg *msg)
 65{
 66	u32 peernode;
 67	u32 orignode;
 68
 69	if (msg_origport(msg) != tipc_port_peerport(p_ptr))
 70		return 0;
 71
 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));
 77}
 78
 79/**
 80 * tipc_port_mcast_xmit - send a multicast message to local and remote
 81 * destinations
 82 */
 83int tipc_port_mcast_xmit(struct tipc_port *oport,
 84			 struct tipc_name_seq const *seq,
 85			 struct iovec const *msg_sect,
 86			 unsigned int len)
 87{
 88	struct tipc_msg *hdr;
 89	struct sk_buff *buf;
 90	struct sk_buff *ibuf = NULL;
 91	struct tipc_port_list dports = {0, NULL, };
 92	int ext_targets;
 93	int res;
 94
 95	/* Create multicast message */
 96	hdr = &oport->phdr;
 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);
106	if (unlikely(!buf))
107		return res;
108
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);
112
113	/* Send message to destinations (duplicate it only if necessary) */
114	if (ext_targets) {
115		if (dports.count != 0) {
116			ibuf = skb_copy(buf, GFP_ATOMIC);
117			if (ibuf == NULL) {
118				tipc_port_list_free(&dports);
119				kfree_skb(buf);
120				return -ENOMEM;
121			}
122		}
123		res = tipc_bclink_xmit(buf);
124		if ((res < 0) && (dports.count != 0))
125			kfree_skb(ibuf);
126	} else {
127		ibuf = buf;
128	}
129
130	if (res >= 0) {
131		if (ibuf)
132			tipc_port_mcast_rcv(ibuf, &dports);
133	} else {
134		tipc_port_list_free(&dports);
135	}
136	return res;
137}
138
139/**
140 * tipc_port_mcast_rcv - deliver multicast message to all destination ports
141 *
142 * If there is no port list, perform a lookup to create one
143 */
144void tipc_port_mcast_rcv(struct sk_buff *buf, struct tipc_port_list *dp)
145{
146	struct tipc_msg *msg;
147	struct tipc_port_list dports = {0, NULL, };
148	struct tipc_port_list *item = dp;
149	int cnt = 0;
150
151	msg = buf_msg(buf);
152
153	/* Create destination port list, if one wasn't supplied */
154	if (dp == NULL) {
155		tipc_nametbl_mc_translate(msg_nametype(msg),
156				     msg_namelower(msg),
157				     msg_nameupper(msg),
158				     TIPC_CLUSTER_SCOPE,
159				     &dports);
160		item = dp = &dports;
161	}
162
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]);
168			tipc_port_rcv(buf);
169			tipc_port_list_free(dp);
170			return;
171		}
172		for (; cnt < dp->count; cnt++) {
173			int index = cnt % PLSIZE;
174			struct sk_buff *b = skb_clone(buf, GFP_ATOMIC);
175
176			if (b == NULL) {
177				pr_warn("Unable to deliver multicast message(s)\n");
178				goto exit;
179			}
180			if ((index == 0) && (cnt != 0))
181				item = item->next;
182			msg_set_destport(buf_msg(b), item->ports[index]);
183			tipc_port_rcv(b);
184		}
185	}
186exit:
187	kfree_skb(buf);
188	tipc_port_list_free(dp);
189}
190
191
192void tipc_port_wakeup(struct tipc_port *port)
193{
194	tipc_sock_wakeup(tipc_port_to_sock(port));
195}
196
197/* tipc_port_init - intiate TIPC port and lock it
198 *
199 * Returns obtained reference if initialization is successful, zero otherwise
200 */
201u32 tipc_port_init(struct tipc_port *p_ptr,
202		   const unsigned int importance)
203{
204	struct tipc_msg *msg;
205	u32 ref;
206
207	ref = tipc_ref_acquire(p_ptr, &p_ptr->lock);
208	if (!ref) {
209		pr_warn("Port registration failed, ref. table exhausted\n");
210		return 0;
211	}
212
213	p_ptr->max_pkt = MAX_PKT_DEFAULT;
214	p_ptr->ref = ref;
215	INIT_LIST_HEAD(&p_ptr->wait_list);
216	INIT_LIST_HEAD(&p_ptr->subscription.nodesub_list);
217	k_init_timer(&p_ptr->timer, (Handler)port_timeout, ref);
218	INIT_LIST_HEAD(&p_ptr->publications);
219	INIT_LIST_HEAD(&p_ptr->port_list);
220
221	/*
222	 * Must hold port list lock while initializing message header template
223	 * to ensure a change to node's own network address doesn't result
224	 * in template containing out-dated network address information
225	 */
226	spin_lock_bh(&tipc_port_list_lock);
227	msg = &p_ptr->phdr;
228	tipc_msg_init(msg, importance, TIPC_NAMED_MSG, NAMED_H_SIZE, 0);
229	msg_set_origport(msg, ref);
230	list_add_tail(&p_ptr->port_list, &ports);
231	spin_unlock_bh(&tipc_port_list_lock);
232	return ref;
233}
234
235void tipc_port_destroy(struct tipc_port *p_ptr)
236{
237	struct sk_buff *buf = NULL;
238
239	tipc_withdraw(p_ptr, 0, NULL);
240
241	spin_lock_bh(p_ptr->lock);
242	tipc_ref_discard(p_ptr->ref);
243	spin_unlock_bh(p_ptr->lock);
244
245	k_cancel_timer(&p_ptr->timer);
246	if (p_ptr->connected) {
247		buf = port_build_peer_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
248		tipc_nodesub_unsubscribe(&p_ptr->subscription);
249	}
250
251	spin_lock_bh(&tipc_port_list_lock);
252	list_del(&p_ptr->port_list);
253	list_del(&p_ptr->wait_list);
254	spin_unlock_bh(&tipc_port_list_lock);
255	k_term_timer(&p_ptr->timer);
256	tipc_net_route_msg(buf);
257}
258
259/*
260 * port_build_proto_msg(): create connection protocol message for port
261 *
262 * On entry the port must be locked and connected.
263 */
264static struct sk_buff *port_build_proto_msg(struct tipc_port *p_ptr,
265					    u32 type, u32 ack)
266{
267	struct sk_buff *buf;
268	struct tipc_msg *msg;
269
270	buf = tipc_buf_acquire(INT_H_SIZE);
271	if (buf) {
272		msg = buf_msg(buf);
273		tipc_msg_init(msg, CONN_MANAGER, type, INT_H_SIZE,
274			      tipc_port_peernode(p_ptr));
275		msg_set_destport(msg, tipc_port_peerport(p_ptr));
276		msg_set_origport(msg, p_ptr->ref);
277		msg_set_msgcnt(msg, ack);
278	}
279	return buf;
280}
281
282int tipc_reject_msg(struct sk_buff *buf, u32 err)
283{
284	struct tipc_msg *msg = buf_msg(buf);
285	struct sk_buff *rbuf;
286	struct tipc_msg *rmsg;
287	int hdr_sz;
288	u32 imp;
289	u32 data_sz = msg_data_sz(msg);
290	u32 src_node;
291	u32 rmsg_sz;
292
293	/* discard rejected message if it shouldn't be returned to sender */
294	if (WARN(!msg_isdata(msg),
295		 "attempt to reject message with user=%u", msg_user(msg))) {
296		dump_stack();
297		goto exit;
298	}
299	if (msg_errcode(msg) || msg_dest_droppable(msg))
300		goto exit;
301
302	/*
303	 * construct returned message by copying rejected message header and
304	 * data (or subset), then updating header fields that need adjusting
305	 */
306	hdr_sz = msg_hdr_sz(msg);
307	rmsg_sz = hdr_sz + min_t(u32, data_sz, MAX_REJECT_SIZE);
308
309	rbuf = tipc_buf_acquire(rmsg_sz);
310	if (rbuf == NULL)
311		goto exit;
312
313	rmsg = buf_msg(rbuf);
314	skb_copy_to_linear_data(rbuf, msg, rmsg_sz);
315
316	if (msg_connected(rmsg)) {
317		imp = msg_importance(rmsg);
318		if (imp < TIPC_CRITICAL_IMPORTANCE)
319			msg_set_importance(rmsg, ++imp);
320	}
321	msg_set_non_seq(rmsg, 0);
322	msg_set_size(rmsg, rmsg_sz);
323	msg_set_errcode(rmsg, err);
324	msg_set_prevnode(rmsg, tipc_own_addr);
325	msg_swap_words(rmsg, 4, 5);
326	if (!msg_short(rmsg))
327		msg_swap_words(rmsg, 6, 7);
328
329	/* send self-abort message when rejecting on a connected port */
330	if (msg_connected(msg)) {
331		struct tipc_port *p_ptr = tipc_port_lock(msg_destport(msg));
332
333		if (p_ptr) {
334			struct sk_buff *abuf = NULL;
335
336			if (p_ptr->connected)
337				abuf = port_build_self_abort_msg(p_ptr, err);
338			tipc_port_unlock(p_ptr);
339			tipc_net_route_msg(abuf);
340		}
341	}
342
343	/* send returned message & dispose of rejected message */
344	src_node = msg_prevnode(msg);
345	if (in_own_node(src_node))
346		tipc_port_rcv(rbuf);
347	else
348		tipc_link_xmit(rbuf, src_node, msg_link_selector(rmsg));
349exit:
350	kfree_skb(buf);
351	return data_sz;
352}
353
354int tipc_port_iovec_reject(struct tipc_port *p_ptr, struct tipc_msg *hdr,
355			   struct iovec const *msg_sect, unsigned int len,
356			   int err)
357{
358	struct sk_buff *buf;
359	int res;
360
361	res = tipc_msg_build(hdr, msg_sect, len, MAX_MSG_SIZE, &buf);
362	if (!buf)
363		return res;
364
365	return tipc_reject_msg(buf, err);
366}
367
368static void port_timeout(unsigned long ref)
369{
370	struct tipc_port *p_ptr = tipc_port_lock(ref);
371	struct sk_buff *buf = NULL;
372
373	if (!p_ptr)
374		return;
375
376	if (!p_ptr->connected) {
377		tipc_port_unlock(p_ptr);
378		return;
379	}
380
381	/* Last probe answered ? */
382	if (p_ptr->probing_state == PROBING) {
383		buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
384	} else {
385		buf = port_build_proto_msg(p_ptr, CONN_PROBE, 0);
386		p_ptr->probing_state = PROBING;
387		k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
388	}
389	tipc_port_unlock(p_ptr);
390	tipc_net_route_msg(buf);
391}
392
393
394static void port_handle_node_down(unsigned long ref)
395{
396	struct tipc_port *p_ptr = tipc_port_lock(ref);
397	struct sk_buff *buf = NULL;
398
399	if (!p_ptr)
400		return;
401	buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_NODE);
402	tipc_port_unlock(p_ptr);
403	tipc_net_route_msg(buf);
404}
405
406
407static struct sk_buff *port_build_self_abort_msg(struct tipc_port *p_ptr, u32 err)
408{
409	struct sk_buff *buf = port_build_peer_abort_msg(p_ptr, err);
410
411	if (buf) {
412		struct tipc_msg *msg = buf_msg(buf);
413		msg_swap_words(msg, 4, 5);
414		msg_swap_words(msg, 6, 7);
415	}
416	return buf;
417}
418
419
420static struct sk_buff *port_build_peer_abort_msg(struct tipc_port *p_ptr, u32 err)
421{
422	struct sk_buff *buf;
423	struct tipc_msg *msg;
424	u32 imp;
425
426	if (!p_ptr->connected)
427		return NULL;
428
429	buf = tipc_buf_acquire(BASIC_H_SIZE);
430	if (buf) {
431		msg = buf_msg(buf);
432		memcpy(msg, &p_ptr->phdr, BASIC_H_SIZE);
433		msg_set_hdr_sz(msg, BASIC_H_SIZE);
434		msg_set_size(msg, BASIC_H_SIZE);
435		imp = msg_importance(msg);
436		if (imp < TIPC_CRITICAL_IMPORTANCE)
437			msg_set_importance(msg, ++imp);
438		msg_set_errcode(msg, err);
439	}
440	return buf;
441}
442
443void tipc_port_proto_rcv(struct sk_buff *buf)
444{
445	struct tipc_msg *msg = buf_msg(buf);
446	struct tipc_port *p_ptr;
447	struct sk_buff *r_buf = NULL;
448	u32 destport = msg_destport(msg);
449	int wakeable;
450
451	/* Validate connection */
452	p_ptr = tipc_port_lock(destport);
453	if (!p_ptr || !p_ptr->connected || !tipc_port_peer_msg(p_ptr, msg)) {
454		r_buf = tipc_buf_acquire(BASIC_H_SIZE);
455		if (r_buf) {
456			msg = buf_msg(r_buf);
457			tipc_msg_init(msg, TIPC_HIGH_IMPORTANCE, TIPC_CONN_MSG,
458				      BASIC_H_SIZE, msg_orignode(msg));
459			msg_set_errcode(msg, TIPC_ERR_NO_PORT);
460			msg_set_origport(msg, destport);
461			msg_set_destport(msg, msg_origport(msg));
462		}
463		if (p_ptr)
464			tipc_port_unlock(p_ptr);
465		goto exit;
466	}
467
468	/* Process protocol message sent by peer */
469	switch (msg_type(msg)) {
470	case CONN_ACK:
471		wakeable = tipc_port_congested(p_ptr) && p_ptr->congested;
472		p_ptr->acked += msg_msgcnt(msg);
473		if (!tipc_port_congested(p_ptr)) {
474			p_ptr->congested = 0;
475			if (wakeable)
476				tipc_port_wakeup(p_ptr);
477		}
478		break;
479	case CONN_PROBE:
480		r_buf = port_build_proto_msg(p_ptr, CONN_PROBE_REPLY, 0);
481		break;
482	default:
483		/* CONN_PROBE_REPLY or unrecognized - no action required */
484		break;
485	}
486	p_ptr->probing_state = CONFIRMED;
487	tipc_port_unlock(p_ptr);
488exit:
489	tipc_net_route_msg(r_buf);
490	kfree_skb(buf);
491}
492
493static int port_print(struct tipc_port *p_ptr, char *buf, int len, int full_id)
494{
495	struct publication *publ;
496	int ret;
497
498	if (full_id)
499		ret = tipc_snprintf(buf, len, "<%u.%u.%u:%u>:",
500				    tipc_zone(tipc_own_addr),
501				    tipc_cluster(tipc_own_addr),
502				    tipc_node(tipc_own_addr), p_ptr->ref);
503	else
504		ret = tipc_snprintf(buf, len, "%-10u:", p_ptr->ref);
505
506	if (p_ptr->connected) {
507		u32 dport = tipc_port_peerport(p_ptr);
508		u32 destnode = tipc_port_peernode(p_ptr);
509
510		ret += tipc_snprintf(buf + ret, len - ret,
511				     " connected to <%u.%u.%u:%u>",
512				     tipc_zone(destnode),
513				     tipc_cluster(destnode),
514				     tipc_node(destnode), dport);
515		if (p_ptr->conn_type != 0)
516			ret += tipc_snprintf(buf + ret, len - ret,
517					     " via {%u,%u}", p_ptr->conn_type,
518					     p_ptr->conn_instance);
519	} else if (p_ptr->published) {
520		ret += tipc_snprintf(buf + ret, len - ret, " bound to");
521		list_for_each_entry(publ, &p_ptr->publications, pport_list) {
522			if (publ->lower == publ->upper)
523				ret += tipc_snprintf(buf + ret, len - ret,
524						     " {%u,%u}", publ->type,
525						     publ->lower);
526			else
527				ret += tipc_snprintf(buf + ret, len - ret,
528						     " {%u,%u,%u}", publ->type,
529						     publ->lower, publ->upper);
530		}
531	}
532	ret += tipc_snprintf(buf + ret, len - ret, "\n");
533	return ret;
534}
535
536struct sk_buff *tipc_port_get_ports(void)
537{
538	struct sk_buff *buf;
539	struct tlv_desc *rep_tlv;
540	char *pb;
541	int pb_len;
542	struct tipc_port *p_ptr;
543	int str_len = 0;
544
545	buf = tipc_cfg_reply_alloc(TLV_SPACE(ULTRA_STRING_MAX_LEN));
546	if (!buf)
547		return NULL;
548	rep_tlv = (struct tlv_desc *)buf->data;
549	pb = TLV_DATA(rep_tlv);
550	pb_len = ULTRA_STRING_MAX_LEN;
551
552	spin_lock_bh(&tipc_port_list_lock);
553	list_for_each_entry(p_ptr, &ports, port_list) {
554		spin_lock_bh(p_ptr->lock);
555		str_len += port_print(p_ptr, pb, pb_len, 0);
556		spin_unlock_bh(p_ptr->lock);
557	}
558	spin_unlock_bh(&tipc_port_list_lock);
559	str_len += 1;	/* for "\0" */
560	skb_put(buf, TLV_SPACE(str_len));
561	TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
562
563	return buf;
564}
565
566void tipc_port_reinit(void)
567{
568	struct tipc_port *p_ptr;
569	struct tipc_msg *msg;
570
571	spin_lock_bh(&tipc_port_list_lock);
572	list_for_each_entry(p_ptr, &ports, port_list) {
573		msg = &p_ptr->phdr;
574		msg_set_prevnode(msg, tipc_own_addr);
575		msg_set_orignode(msg, tipc_own_addr);
576	}
577	spin_unlock_bh(&tipc_port_list_lock);
578}
579
580void tipc_acknowledge(u32 ref, u32 ack)
581{
582	struct tipc_port *p_ptr;
583	struct sk_buff *buf = NULL;
584
585	p_ptr = tipc_port_lock(ref);
586	if (!p_ptr)
587		return;
588	if (p_ptr->connected) {
589		p_ptr->conn_unacked -= ack;
590		buf = port_build_proto_msg(p_ptr, CONN_ACK, ack);
591	}
592	tipc_port_unlock(p_ptr);
593	tipc_net_route_msg(buf);
594}
595
596int tipc_publish(struct tipc_port *p_ptr, unsigned int scope,
597		 struct tipc_name_seq const *seq)
598{
599	struct publication *publ;
600	u32 key;
601
602	if (p_ptr->connected)
603		return -EINVAL;
604	key = p_ptr->ref + p_ptr->pub_count + 1;
605	if (key == p_ptr->ref)
606		return -EADDRINUSE;
607
608	publ = tipc_nametbl_publish(seq->type, seq->lower, seq->upper,
609				    scope, p_ptr->ref, key);
610	if (publ) {
611		list_add(&publ->pport_list, &p_ptr->publications);
612		p_ptr->pub_count++;
613		p_ptr->published = 1;
614		return 0;
615	}
616	return -EINVAL;
617}
618
619int tipc_withdraw(struct tipc_port *p_ptr, unsigned int scope,
620		  struct tipc_name_seq const *seq)
621{
622	struct publication *publ;
623	struct publication *tpubl;
624	int res = -EINVAL;
625
626	if (!seq) {
627		list_for_each_entry_safe(publ, tpubl,
628					 &p_ptr->publications, pport_list) {
629			tipc_nametbl_withdraw(publ->type, publ->lower,
630					      publ->ref, publ->key);
631		}
632		res = 0;
633	} else {
634		list_for_each_entry_safe(publ, tpubl,
635					 &p_ptr->publications, pport_list) {
636			if (publ->scope != scope)
637				continue;
638			if (publ->type != seq->type)
639				continue;
640			if (publ->lower != seq->lower)
641				continue;
642			if (publ->upper != seq->upper)
643				break;
644			tipc_nametbl_withdraw(publ->type, publ->lower,
645					      publ->ref, publ->key);
646			res = 0;
647			break;
648		}
649	}
650	if (list_empty(&p_ptr->publications))
651		p_ptr->published = 0;
652	return res;
653}
654
655int tipc_port_connect(u32 ref, struct tipc_portid const *peer)
656{
657	struct tipc_port *p_ptr;
658	int res;
659
660	p_ptr = tipc_port_lock(ref);
661	if (!p_ptr)
662		return -EINVAL;
663	res = __tipc_port_connect(ref, p_ptr, peer);
664	tipc_port_unlock(p_ptr);
665	return res;
666}
667
668/*
669 * __tipc_port_connect - connect to a remote peer
670 *
671 * Port must be locked.
672 */
673int __tipc_port_connect(u32 ref, struct tipc_port *p_ptr,
674			struct tipc_portid const *peer)
675{
676	struct tipc_msg *msg;
677	int res = -EINVAL;
678
679	if (p_ptr->published || p_ptr->connected)
680		goto exit;
681	if (!peer->ref)
682		goto exit;
683
684	msg = &p_ptr->phdr;
685	msg_set_destnode(msg, peer->node);
686	msg_set_destport(msg, peer->ref);
687	msg_set_type(msg, TIPC_CONN_MSG);
688	msg_set_lookup_scope(msg, 0);
689	msg_set_hdr_sz(msg, SHORT_H_SIZE);
690
691	p_ptr->probing_interval = PROBING_INTERVAL;
692	p_ptr->probing_state = CONFIRMED;
693	p_ptr->connected = 1;
694	k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
695
696	tipc_nodesub_subscribe(&p_ptr->subscription, peer->node,
697			  (void *)(unsigned long)ref,
698			  (net_ev_handler)port_handle_node_down);
699	res = 0;
700exit:
701	p_ptr->max_pkt = tipc_link_get_max_pkt(peer->node, ref);
702	return res;
703}
704
705/*
706 * __tipc_disconnect - disconnect port from peer
707 *
708 * Port must be locked.
709 */
710int __tipc_port_disconnect(struct tipc_port *tp_ptr)
711{
712	if (tp_ptr->connected) {
713		tp_ptr->connected = 0;
714		/* let timer expire on it's own to avoid deadlock! */
715		tipc_nodesub_unsubscribe(&tp_ptr->subscription);
716		return 0;
717	}
718
719	return -ENOTCONN;
720}
721
722/*
723 * tipc_port_disconnect(): Disconnect port form peer.
724 *                    This is a node local operation.
725 */
726int tipc_port_disconnect(u32 ref)
727{
728	struct tipc_port *p_ptr;
729	int res;
730
731	p_ptr = tipc_port_lock(ref);
732	if (!p_ptr)
733		return -EINVAL;
734	res = __tipc_port_disconnect(p_ptr);
735	tipc_port_unlock(p_ptr);
736	return res;
737}
738
739/*
740 * tipc_port_shutdown(): Send a SHUTDOWN msg to peer and disconnect
741 */
742int tipc_port_shutdown(u32 ref)
743{
744	struct tipc_port *p_ptr;
745	struct sk_buff *buf = NULL;
746
747	p_ptr = tipc_port_lock(ref);
748	if (!p_ptr)
749		return -EINVAL;
750
751	buf = port_build_peer_abort_msg(p_ptr, TIPC_CONN_SHUTDOWN);
752	tipc_port_unlock(p_ptr);
753	tipc_net_route_msg(buf);
754	return tipc_port_disconnect(ref);
755}
756
757/**
758 * tipc_port_rcv - receive message from lower layer and deliver to port user
759 */
760int tipc_port_rcv(struct sk_buff *buf)
761{
762	struct tipc_port *p_ptr;
763	struct tipc_msg *msg = buf_msg(buf);
764	u32 destport = msg_destport(msg);
765	u32 dsz = msg_data_sz(msg);
766	u32 err;
767
768	/* forward unresolved named message */
769	if (unlikely(!destport)) {
770		tipc_net_route_msg(buf);
771		return dsz;
772	}
773
774	/* validate destination & pass to port, otherwise reject message */
775	p_ptr = tipc_port_lock(destport);
776	if (likely(p_ptr)) {
777		err = tipc_sk_rcv(&tipc_port_to_sock(p_ptr)->sk, buf);
778		tipc_port_unlock(p_ptr);
779		if (likely(!err))
780			return dsz;
781	} else {
782		err = TIPC_ERR_NO_PORT;
783	}
784
785	return tipc_reject_msg(buf, err);
786}
787
788/*
789 *  tipc_port_iovec_rcv: Concatenate and deliver sectioned
790 *                       message for this node.
791 */
792static int tipc_port_iovec_rcv(struct tipc_port *sender,
793			       struct iovec const *msg_sect,
794			       unsigned int len)
795{
796	struct sk_buff *buf;
797	int res;
798
799	res = tipc_msg_build(&sender->phdr, msg_sect, len, MAX_MSG_SIZE, &buf);
800	if (likely(buf))
801		tipc_port_rcv(buf);
802	return res;
803}
804
805/**
806 * tipc_send - send message sections on connection
807 */
808int tipc_send(struct tipc_port *p_ptr,
809	      struct iovec const *msg_sect,
810	      unsigned int len)
811{
812	u32 destnode;
813	int res;
814
815	if (!p_ptr->connected)
816		return -EINVAL;
817
818	p_ptr->congested = 1;
819	if (!tipc_port_congested(p_ptr)) {
820		destnode = tipc_port_peernode(p_ptr);
821		if (likely(!in_own_node(destnode)))
822			res = tipc_link_iovec_xmit_fast(p_ptr, msg_sect, len,
823							destnode);
824		else
825			res = tipc_port_iovec_rcv(p_ptr, msg_sect, len);
826
827		if (likely(res != -ELINKCONG)) {
828			p_ptr->congested = 0;
829			if (res > 0)
830				p_ptr->sent++;
831			return res;
832		}
833	}
834	if (tipc_port_unreliable(p_ptr)) {
835		p_ptr->congested = 0;
836		return len;
837	}
838	return -ELINKCONG;
839}
840
841/**
842 * tipc_send2name - send message sections to port name
843 */
844int tipc_send2name(struct tipc_port *p_ptr,
845		   struct tipc_name const *name,
846		   unsigned int domain,
847		   struct iovec const *msg_sect,
848		   unsigned int len)
849{
850	struct tipc_msg *msg;
851	u32 destnode = domain;
852	u32 destport;
853	int res;
854
855	if (p_ptr->connected)
856		return -EINVAL;
857
858	msg = &p_ptr->phdr;
859	msg_set_type(msg, TIPC_NAMED_MSG);
860	msg_set_hdr_sz(msg, NAMED_H_SIZE);
861	msg_set_nametype(msg, name->type);
862	msg_set_nameinst(msg, name->instance);
863	msg_set_lookup_scope(msg, tipc_addr_scope(domain));
864	destport = tipc_nametbl_translate(name->type, name->instance, &destnode);
865	msg_set_destnode(msg, destnode);
866	msg_set_destport(msg, destport);
867
868	if (likely(destport || destnode)) {
869		if (likely(in_own_node(destnode)))
870			res = tipc_port_iovec_rcv(p_ptr, msg_sect, len);
871		else if (tipc_own_addr)
872			res = tipc_link_iovec_xmit_fast(p_ptr, msg_sect, len,
873							destnode);
874		else
875			res = tipc_port_iovec_reject(p_ptr, msg, msg_sect,
876						     len, TIPC_ERR_NO_NODE);
877		if (likely(res != -ELINKCONG)) {
878			if (res > 0)
879				p_ptr->sent++;
880			return res;
881		}
882		if (tipc_port_unreliable(p_ptr))
883			return len;
884
885		return -ELINKCONG;
886	}
887	return tipc_port_iovec_reject(p_ptr, msg, msg_sect, len,
888				      TIPC_ERR_NO_NAME);
889}
890
891/**
892 * tipc_send2port - send message sections to port identity
893 */
894int tipc_send2port(struct tipc_port *p_ptr,
895		   struct tipc_portid const *dest,
896		   struct iovec const *msg_sect,
897		   unsigned int len)
898{
899	struct tipc_msg *msg;
900	int res;
901
902	if (p_ptr->connected)
903		return -EINVAL;
904
905	msg = &p_ptr->phdr;
906	msg_set_type(msg, TIPC_DIRECT_MSG);
907	msg_set_lookup_scope(msg, 0);
908	msg_set_destnode(msg, dest->node);
909	msg_set_destport(msg, dest->ref);
910	msg_set_hdr_sz(msg, BASIC_H_SIZE);
911
912	if (in_own_node(dest->node))
913		res =  tipc_port_iovec_rcv(p_ptr, msg_sect, len);
914	else if (tipc_own_addr)
915		res = tipc_link_iovec_xmit_fast(p_ptr, msg_sect, len,
916						dest->node);
917	else
918		res = tipc_port_iovec_reject(p_ptr, msg, msg_sect, len,
919						TIPC_ERR_NO_NODE);
920	if (likely(res != -ELINKCONG)) {
921		if (res > 0)
922			p_ptr->sent++;
923		return res;
924	}
925	if (tipc_port_unreliable(p_ptr))
926		return len;
927
928	return -ELINKCONG;
929}