Kernel

Kernel Git Source Tree

Root/net/tipc/port.c

1	/*
2	* net/tipc/port.c: TIPC port code
3	*
4	* Copyright (c) 1992-2007, Ericsson AB
5	* Copyright (c) 2004-2008, 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 "dbg.h"
40	#include "port.h"
41	#include "addr.h"
42	#include "link.h"
43	#include "node.h"
44	#include "name_table.h"
45	#include "user_reg.h"
46	#include "msg.h"
47	#include "bcast.h"
48
49	/* Connection management: */
50	#define PROBING_INTERVAL 3600000 /* [ms] => 1 h */
51	#define CONFIRMED 0
52	#define PROBING 1
53
54	#define MAX_REJECT_SIZE 1024
55
56	static struct sk_buff *msg_queue_head = NULL;
57	static struct sk_buff *msg_queue_tail = NULL;
58
59	DEFINE_SPINLOCK(tipc_port_list_lock);
60	static DEFINE_SPINLOCK(queue_lock);
61
62	static LIST_HEAD(ports);
63	static void port_handle_node_down(unsigned long ref);
64	static struct sk_buff* port_build_self_abort_msg(struct port *,u32 err);
65	static struct sk_buff* port_build_peer_abort_msg(struct port *,u32 err);
66	static void port_timeout(unsigned long ref);
67
68
69	static u32 port_peernode(struct port *p_ptr)
70	{
71	return msg_destnode(&p_ptr->publ.phdr);
72	}
73
74	static u32 port_peerport(struct port *p_ptr)
75	{
76	return msg_destport(&p_ptr->publ.phdr);
77	}
78
79	static u32 port_out_seqno(struct port *p_ptr)
80	{
81	return msg_transp_seqno(&p_ptr->publ.phdr);
82	}
83
84	static void port_incr_out_seqno(struct port *p_ptr)
85	{
86	struct tipc_msg *m = &p_ptr->publ.phdr;
87
88	if (likely(!msg_routed(m)))
89	return;
90	msg_set_transp_seqno(m, (msg_transp_seqno(m) + 1));
91	}
92
93	/**
94	* tipc_multicast - send a multicast message to local and remote destinations
95	*/
96
97	int tipc_multicast(u32 ref, struct tipc_name_seq const *seq, u32 domain,
98	u32 num_sect, struct iovec const *msg_sect)
99	{
100	struct tipc_msg *hdr;
101	struct sk_buff *buf;
102	struct sk_buff *ibuf = NULL;
103	struct port_list dports = {0, NULL, };
104	struct port *oport = tipc_port_deref(ref);
105	int ext_targets;
106	int res;
107
108	if (unlikely(!oport))
109	return -EINVAL;
110
111	/* Create multicast message */
112
113	hdr = &oport->publ.phdr;
114	msg_set_type(hdr, TIPC_MCAST_MSG);
115	msg_set_nametype(hdr, seq->type);
116	msg_set_namelower(hdr, seq->lower);
117	msg_set_nameupper(hdr, seq->upper);
118	msg_set_hdr_sz(hdr, MCAST_H_SIZE);
119	res = tipc_msg_build(hdr, msg_sect, num_sect, MAX_MSG_SIZE,
120	!oport->user_port, &buf);
121	if (unlikely(!buf))
122	return res;
123
124	/* Figure out where to send multicast message */
125
126	ext_targets = tipc_nametbl_mc_translate(seq->type, seq->lower, seq->upper,
127	TIPC_NODE_SCOPE, &dports);
128
129	/* Send message to destinations (duplicate it only if necessary) */
130
131	if (ext_targets) {
132	if (dports.count != 0) {
133	ibuf = skb_copy(buf, GFP_ATOMIC);
134	if (ibuf == NULL) {
135	tipc_port_list_free(&dports);
136	buf_discard(buf);
137	return -ENOMEM;
138	}
139	}
140	res = tipc_bclink_send_msg(buf);
141	if ((res < 0) && (dports.count != 0)) {
142	buf_discard(ibuf);
143	}
144	} else {
145	ibuf = buf;
146	}
147
148	if (res >= 0) {
149	if (ibuf)
150	tipc_port_recv_mcast(ibuf, &dports);
151	} else {
152	tipc_port_list_free(&dports);
153	}
154	return res;
155	}
156
157	/**
158	* tipc_port_recv_mcast - deliver multicast message to all destination ports
159	*
160	* If there is no port list, perform a lookup to create one
161	*/
162
163	void tipc_port_recv_mcast(struct sk_buff buf, struct port_list dp)
164	{
165	struct tipc_msg* msg;
166	struct port_list dports = {0, NULL, };
167	struct port_list *item = dp;
168	int cnt = 0;
169
170	msg = buf_msg(buf);
171
172	/* Create destination port list, if one wasn't supplied */
173
174	if (dp == NULL) {
175	tipc_nametbl_mc_translate(msg_nametype(msg),
176	msg_namelower(msg),
177	msg_nameupper(msg),
178	TIPC_CLUSTER_SCOPE,
179	&dports);
180	item = dp = &dports;
181	}
182
183	/* Deliver a copy of message to each destination port */
184
185	if (dp->count != 0) {
186	if (dp->count == 1) {
187	msg_set_destport(msg, dp->ports[0]);
188	tipc_port_recv_msg(buf);
189	tipc_port_list_free(dp);
190	return;
191	}
192	for (; cnt < dp->count; cnt++) {
193	int index = cnt % PLSIZE;
194	struct sk_buff *b = skb_clone(buf, GFP_ATOMIC);
195
196	if (b == NULL) {
197	warn("Unable to deliver multicast message(s)\n");
198	msg_dbg(msg, "LOST:");
199	goto exit;
200	}
201	if ((index == 0) && (cnt != 0)) {
202	item = item->next;
203	}
204	msg_set_destport(buf_msg(b),item->ports[index]);
205	tipc_port_recv_msg(b);
206	}
207	}
208	exit:
209	buf_discard(buf);
210	tipc_port_list_free(dp);
211	}
212
213	/**
214	* tipc_createport_raw - create a generic TIPC port
215	*
216	* Returns pointer to (locked) TIPC port, or NULL if unable to create it
217	*/
218
219	struct tipc_port tipc_createport_raw(void usr_handle,
220	u32 (dispatcher)(struct tipc_port , struct sk_buff *),
221	void (wakeup)(struct tipc_port ),
222	const u32 importance)
223	{
224	struct port *p_ptr;
225	struct tipc_msg *msg;
226	u32 ref;
227
228	p_ptr = kzalloc(sizeof(*p_ptr), GFP_ATOMIC);
229	if (!p_ptr) {
230	warn("Port creation failed, no memory\n");
231	return NULL;
232	}
233	ref = tipc_ref_acquire(p_ptr, &p_ptr->publ.lock);
234	if (!ref) {
235	warn("Port creation failed, reference table exhausted\n");
236	kfree(p_ptr);
237	return NULL;
238	}
239
240	p_ptr->publ.usr_handle = usr_handle;
241	p_ptr->publ.max_pkt = MAX_PKT_DEFAULT;
242	p_ptr->publ.ref = ref;
243	msg = &p_ptr->publ.phdr;
244	tipc_msg_init(msg, importance, TIPC_NAMED_MSG, LONG_H_SIZE, 0);
245	msg_set_origport(msg, ref);
246	p_ptr->last_in_seqno = 41;
247	p_ptr->sent = 1;
248	INIT_LIST_HEAD(&p_ptr->wait_list);
249	INIT_LIST_HEAD(&p_ptr->subscription.nodesub_list);
250	p_ptr->dispatcher = dispatcher;
251	p_ptr->wakeup = wakeup;
252	p_ptr->user_port = NULL;
253	k_init_timer(&p_ptr->timer, (Handler)port_timeout, ref);
254	spin_lock_bh(&tipc_port_list_lock);
255	INIT_LIST_HEAD(&p_ptr->publications);
256	INIT_LIST_HEAD(&p_ptr->port_list);
257	list_add_tail(&p_ptr->port_list, &ports);
258	spin_unlock_bh(&tipc_port_list_lock);
259	return &(p_ptr->publ);
260	}
261
262	int tipc_deleteport(u32 ref)
263	{
264	struct port *p_ptr;
265	struct sk_buff *buf = NULL;
266
267	tipc_withdraw(ref, 0, NULL);
268	p_ptr = tipc_port_lock(ref);
269	if (!p_ptr)
270	return -EINVAL;
271
272	tipc_ref_discard(ref);
273	tipc_port_unlock(p_ptr);
274
275	k_cancel_timer(&p_ptr->timer);
276	if (p_ptr->publ.connected) {
277	buf = port_build_peer_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
278	tipc_nodesub_unsubscribe(&p_ptr->subscription);
279	}
280	if (p_ptr->user_port) {
281	tipc_reg_remove_port(p_ptr->user_port);
282	kfree(p_ptr->user_port);
283	}
284
285	spin_lock_bh(&tipc_port_list_lock);
286	list_del(&p_ptr->port_list);
287	list_del(&p_ptr->wait_list);
288	spin_unlock_bh(&tipc_port_list_lock);
289	k_term_timer(&p_ptr->timer);
290	kfree(p_ptr);
291	dbg("Deleted port %u\n", ref);
292	tipc_net_route_msg(buf);
293	return 0;
294	}
295
296	/**
297	* tipc_get_port() - return port associated with 'ref'
298	*
299	* Note: Port is not locked.
300	*/
301
302	struct tipc_port *tipc_get_port(const u32 ref)
303	{
304	return (struct tipc_port *)tipc_ref_deref(ref);
305	}
306
307	/**
308	* tipc_get_handle - return user handle associated to port 'ref'
309	*/
310
311	void *tipc_get_handle(const u32 ref)
312	{
313	struct port *p_ptr;
314	void * handle;
315
316	p_ptr = tipc_port_lock(ref);
317	if (!p_ptr)
318	return NULL;
319	handle = p_ptr->publ.usr_handle;
320	tipc_port_unlock(p_ptr);
321	return handle;
322	}
323
324	static int port_unreliable(struct port *p_ptr)
325	{
326	return msg_src_droppable(&p_ptr->publ.phdr);
327	}
328
329	int tipc_portunreliable(u32 ref, unsigned int *isunreliable)
330	{
331	struct port *p_ptr;
332
333	p_ptr = tipc_port_lock(ref);
334	if (!p_ptr)
335	return -EINVAL;
336	*isunreliable = port_unreliable(p_ptr);
337	tipc_port_unlock(p_ptr);
338	return 0;
339	}
340
341	int tipc_set_portunreliable(u32 ref, unsigned int isunreliable)
342	{
343	struct port *p_ptr;
344
345	p_ptr = tipc_port_lock(ref);
346	if (!p_ptr)
347	return -EINVAL;
348	msg_set_src_droppable(&p_ptr->publ.phdr, (isunreliable != 0));
349	tipc_port_unlock(p_ptr);
350	return 0;
351	}
352
353	static int port_unreturnable(struct port *p_ptr)
354	{
355	return msg_dest_droppable(&p_ptr->publ.phdr);
356	}
357
358	int tipc_portunreturnable(u32 ref, unsigned int *isunrejectable)
359	{
360	struct port *p_ptr;
361
362	p_ptr = tipc_port_lock(ref);
363	if (!p_ptr)
364	return -EINVAL;
365	*isunrejectable = port_unreturnable(p_ptr);
366	tipc_port_unlock(p_ptr);
367	return 0;
368	}
369
370	int tipc_set_portunreturnable(u32 ref, unsigned int isunrejectable)
371	{
372	struct port *p_ptr;
373
374	p_ptr = tipc_port_lock(ref);
375	if (!p_ptr)
376	return -EINVAL;
377	msg_set_dest_droppable(&p_ptr->publ.phdr, (isunrejectable != 0));
378	tipc_port_unlock(p_ptr);
379	return 0;
380	}
381
382	/*
383	* port_build_proto_msg(): build a port level protocol
384	* or a connection abortion message. Called with
385	* tipc_port lock on.
386	*/
387	static struct sk_buff *port_build_proto_msg(u32 destport, u32 destnode,
388	u32 origport, u32 orignode,
389	u32 usr, u32 type, u32 err,
390	u32 seqno, u32 ack)
391	{
392	struct sk_buff *buf;
393	struct tipc_msg *msg;
394
395	buf = buf_acquire(LONG_H_SIZE);
396	if (buf) {
397	msg = buf_msg(buf);
398	tipc_msg_init(msg, usr, type, LONG_H_SIZE, destnode);
399	msg_set_errcode(msg, err);
400	msg_set_destport(msg, destport);
401	msg_set_origport(msg, origport);
402	msg_set_orignode(msg, orignode);
403	msg_set_transp_seqno(msg, seqno);
404	msg_set_msgcnt(msg, ack);
405	msg_dbg(msg, "PORT>SEND>:");
406	}
407	return buf;
408	}
409
410	int tipc_reject_msg(struct sk_buff *buf, u32 err)
411	{
412	struct tipc_msg *msg = buf_msg(buf);
413	struct sk_buff *rbuf;
414	struct tipc_msg *rmsg;
415	int hdr_sz;
416	u32 imp = msg_importance(msg);
417	u32 data_sz = msg_data_sz(msg);
418
419	if (data_sz > MAX_REJECT_SIZE)
420	data_sz = MAX_REJECT_SIZE;
421	if (msg_connected(msg) && (imp < TIPC_CRITICAL_IMPORTANCE))
422	imp++;
423	msg_dbg(msg, "port->rej: ");
424
425	/* discard rejected message if it shouldn't be returned to sender */
426	if (msg_errcode(msg) \|\| msg_dest_droppable(msg)) {
427	buf_discard(buf);
428	return data_sz;
429	}
430
431	/* construct rejected message */
432	if (msg_mcast(msg))
433	hdr_sz = MCAST_H_SIZE;
434	else
435	hdr_sz = LONG_H_SIZE;
436	rbuf = buf_acquire(data_sz + hdr_sz);
437	if (rbuf == NULL) {
438	buf_discard(buf);
439	return data_sz;
440	}
441	rmsg = buf_msg(rbuf);
442	tipc_msg_init(rmsg, imp, msg_type(msg), hdr_sz, msg_orignode(msg));
443	msg_set_errcode(rmsg, err);
444	msg_set_destport(rmsg, msg_origport(msg));
445	msg_set_origport(rmsg, msg_destport(msg));
446	if (msg_short(msg)) {
447	msg_set_orignode(rmsg, tipc_own_addr);
448	/* leave name type & instance as zeroes */
449	} else {
450	msg_set_orignode(rmsg, msg_destnode(msg));
451	msg_set_nametype(rmsg, msg_nametype(msg));
452	msg_set_nameinst(rmsg, msg_nameinst(msg));
453	}
454	msg_set_size(rmsg, data_sz + hdr_sz);
455	skb_copy_to_linear_data_offset(rbuf, hdr_sz, msg_data(msg), data_sz);
456
457	/* send self-abort message when rejecting on a connected port */
458	if (msg_connected(msg)) {
459	struct sk_buff *abuf = NULL;
460	struct port *p_ptr = tipc_port_lock(msg_destport(msg));
461
462	if (p_ptr) {
463	if (p_ptr->publ.connected)
464	abuf = port_build_self_abort_msg(p_ptr, err);
465	tipc_port_unlock(p_ptr);
466	}
467	tipc_net_route_msg(abuf);
468	}
469
470	/* send rejected message */
471	buf_discard(buf);
472	tipc_net_route_msg(rbuf);
473	return data_sz;
474	}
475
476	int tipc_port_reject_sections(struct port p_ptr, struct tipc_msg hdr,
477	struct iovec const *msg_sect, u32 num_sect,
478	int err)
479	{
480	struct sk_buff *buf;
481	int res;
482
483	res = tipc_msg_build(hdr, msg_sect, num_sect, MAX_MSG_SIZE,
484	!p_ptr->user_port, &buf);
485	if (!buf)
486	return res;
487
488	return tipc_reject_msg(buf, err);
489	}
490
491	static void port_timeout(unsigned long ref)
492	{
493	struct port *p_ptr = tipc_port_lock(ref);
494	struct sk_buff *buf = NULL;
495
496	if (!p_ptr)
497	return;
498
499	if (!p_ptr->publ.connected) {
500	tipc_port_unlock(p_ptr);
501	return;
502	}
503
504	/* Last probe answered ? */
505	if (p_ptr->probing_state == PROBING) {
506	buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
507	} else {
508	buf = port_build_proto_msg(port_peerport(p_ptr),
509	port_peernode(p_ptr),
510	p_ptr->publ.ref,
511	tipc_own_addr,
512	CONN_MANAGER,
513	CONN_PROBE,
514	TIPC_OK,
515	port_out_seqno(p_ptr),
516	0);
517	port_incr_out_seqno(p_ptr);
518	p_ptr->probing_state = PROBING;
519	k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
520	}
521	tipc_port_unlock(p_ptr);
522	tipc_net_route_msg(buf);
523	}
524
525
526	static void port_handle_node_down(unsigned long ref)
527	{
528	struct port *p_ptr = tipc_port_lock(ref);
529	struct sk_buff* buf = NULL;
530
531	if (!p_ptr)
532	return;
533	buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_NODE);
534	tipc_port_unlock(p_ptr);
535	tipc_net_route_msg(buf);
536	}
537
538
539	static struct sk_buff port_build_self_abort_msg(struct port p_ptr, u32 err)
540	{
541	u32 imp = msg_importance(&p_ptr->publ.phdr);
542
543	if (!p_ptr->publ.connected)
544	return NULL;
545	if (imp < TIPC_CRITICAL_IMPORTANCE)
546	imp++;
547	return port_build_proto_msg(p_ptr->publ.ref,
548	tipc_own_addr,
549	port_peerport(p_ptr),
550	port_peernode(p_ptr),
551	imp,
552	TIPC_CONN_MSG,
553	err,
554	p_ptr->last_in_seqno + 1,
555	0);
556	}
557
558
559	static struct sk_buff port_build_peer_abort_msg(struct port p_ptr, u32 err)
560	{
561	u32 imp = msg_importance(&p_ptr->publ.phdr);
562
563	if (!p_ptr->publ.connected)
564	return NULL;
565	if (imp < TIPC_CRITICAL_IMPORTANCE)
566	imp++;
567	return port_build_proto_msg(port_peerport(p_ptr),
568	port_peernode(p_ptr),
569	p_ptr->publ.ref,
570	tipc_own_addr,
571	imp,
572	TIPC_CONN_MSG,
573	err,
574	port_out_seqno(p_ptr),
575	0);
576	}
577
578	void tipc_port_recv_proto_msg(struct sk_buff *buf)
579	{
580	struct tipc_msg *msg = buf_msg(buf);
581	struct port *p_ptr = tipc_port_lock(msg_destport(msg));
582	u32 err = TIPC_OK;
583	struct sk_buff *r_buf = NULL;
584	struct sk_buff *abort_buf = NULL;
585
586	msg_dbg(msg, "PORT<RECV<:");
587
588	if (!p_ptr) {
589	err = TIPC_ERR_NO_PORT;
590	} else if (p_ptr->publ.connected) {
591	if (port_peernode(p_ptr) != msg_orignode(msg))
592	err = TIPC_ERR_NO_PORT;
593	if (port_peerport(p_ptr) != msg_origport(msg))
594	err = TIPC_ERR_NO_PORT;
595	if (!err && msg_routed(msg)) {
596	u32 seqno = msg_transp_seqno(msg);
597	u32 myno = ++p_ptr->last_in_seqno;
598	if (seqno != myno) {
599	err = TIPC_ERR_NO_PORT;
600	abort_buf = port_build_self_abort_msg(p_ptr, err);
601	}
602	}
603	if (msg_type(msg) == CONN_ACK) {
604	int wakeup = tipc_port_congested(p_ptr) &&
605	p_ptr->publ.congested &&
606	p_ptr->wakeup;
607	p_ptr->acked += msg_msgcnt(msg);
608	if (tipc_port_congested(p_ptr))
609	goto exit;
610	p_ptr->publ.congested = 0;
611	if (!wakeup)
612	goto exit;
613	p_ptr->wakeup(&p_ptr->publ);
614	goto exit;
615	}
616	} else if (p_ptr->publ.published) {
617	err = TIPC_ERR_NO_PORT;
618	}
619	if (err) {
620	r_buf = port_build_proto_msg(msg_origport(msg),
621	msg_orignode(msg),
622	msg_destport(msg),
623	tipc_own_addr,
624	TIPC_HIGH_IMPORTANCE,
625	TIPC_CONN_MSG,
626	err,
627	0,
628	0);
629	goto exit;
630	}
631
632	/* All is fine */
633	if (msg_type(msg) == CONN_PROBE) {
634	r_buf = port_build_proto_msg(msg_origport(msg),
635	msg_orignode(msg),
636	msg_destport(msg),
637	tipc_own_addr,
638	CONN_MANAGER,
639	CONN_PROBE_REPLY,
640	TIPC_OK,
641	port_out_seqno(p_ptr),
642	0);
643	}
644	p_ptr->probing_state = CONFIRMED;
645	port_incr_out_seqno(p_ptr);
646	exit:
647	if (p_ptr)
648	tipc_port_unlock(p_ptr);
649	tipc_net_route_msg(r_buf);
650	tipc_net_route_msg(abort_buf);
651	buf_discard(buf);
652	}
653
654	static void port_print(struct port p_ptr, struct print_buf buf, int full_id)
655	{
656	struct publication *publ;
657
658	if (full_id)
659	tipc_printf(buf, "<%u.%u.%u:%u>:",
660	tipc_zone(tipc_own_addr), tipc_cluster(tipc_own_addr),
661	tipc_node(tipc_own_addr), p_ptr->publ.ref);
662	else
663	tipc_printf(buf, "%-10u:", p_ptr->publ.ref);
664
665	if (p_ptr->publ.connected) {
666	u32 dport = port_peerport(p_ptr);
667	u32 destnode = port_peernode(p_ptr);
668
669	tipc_printf(buf, " connected to <%u.%u.%u:%u>",
670	tipc_zone(destnode), tipc_cluster(destnode),
671	tipc_node(destnode), dport);
672	if (p_ptr->publ.conn_type != 0)
673	tipc_printf(buf, " via {%u,%u}",
674	p_ptr->publ.conn_type,
675	p_ptr->publ.conn_instance);
676	}
677	else if (p_ptr->publ.published) {
678	tipc_printf(buf, " bound to");
679	list_for_each_entry(publ, &p_ptr->publications, pport_list) {
680	if (publ->lower == publ->upper)
681	tipc_printf(buf, " {%u,%u}", publ->type,
682	publ->lower);
683	else
684	tipc_printf(buf, " {%u,%u,%u}", publ->type,
685	publ->lower, publ->upper);
686	}
687	}
688	tipc_printf(buf, "\n");
689	}
690
691	#define MAX_PORT_QUERY 32768
692
693	struct sk_buff *tipc_port_get_ports(void)
694	{
695	struct sk_buff *buf;
696	struct tlv_desc *rep_tlv;
697	struct print_buf pb;
698	struct port *p_ptr;
699	int str_len;
700
701	buf = tipc_cfg_reply_alloc(TLV_SPACE(MAX_PORT_QUERY));
702	if (!buf)
703	return NULL;
704	rep_tlv = (struct tlv_desc *)buf->data;
705
706	tipc_printbuf_init(&pb, TLV_DATA(rep_tlv), MAX_PORT_QUERY);
707	spin_lock_bh(&tipc_port_list_lock);
708	list_for_each_entry(p_ptr, &ports, port_list) {
709	spin_lock_bh(p_ptr->publ.lock);
710	port_print(p_ptr, &pb, 0);
711	spin_unlock_bh(p_ptr->publ.lock);
712	}
713	spin_unlock_bh(&tipc_port_list_lock);
714	str_len = tipc_printbuf_validate(&pb);
715
716	skb_put(buf, TLV_SPACE(str_len));
717	TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
718
719	return buf;
720	}
721
722	#if 0
723
724	#define MAX_PORT_STATS 2000
725
726	struct sk_buff port_show_stats(const void req_tlv_area, int req_tlv_space)
727	{
728	u32 ref;
729	struct port *p_ptr;
730	struct sk_buff *buf;
731	struct tlv_desc *rep_tlv;
732	struct print_buf pb;
733	int str_len;
734
735	if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_PORT_REF))
736	return cfg_reply_error_string(TIPC_CFG_TLV_ERROR);
737
738	ref = (u32 )TLV_DATA(req_tlv_area);
739	ref = ntohl(ref);
740
741	p_ptr = tipc_port_lock(ref);
742	if (!p_ptr)
743	return cfg_reply_error_string("port not found");
744
745	buf = tipc_cfg_reply_alloc(TLV_SPACE(MAX_PORT_STATS));
746	if (!buf) {
747	tipc_port_unlock(p_ptr);
748	return NULL;
749	}
750	rep_tlv = (struct tlv_desc *)buf->data;
751
752	tipc_printbuf_init(&pb, TLV_DATA(rep_tlv), MAX_PORT_STATS);
753	port_print(p_ptr, &pb, 1);
754	/* NEED TO FILL IN ADDITIONAL PORT STATISTICS HERE */
755	tipc_port_unlock(p_ptr);
756	str_len = tipc_printbuf_validate(&pb);
757
758	skb_put(buf, TLV_SPACE(str_len));
759	TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
760
761	return buf;
762	}
763
764	#endif
765
766	void tipc_port_reinit(void)
767	{
768	struct port *p_ptr;
769	struct tipc_msg *msg;
770
771	spin_lock_bh(&tipc_port_list_lock);
772	list_for_each_entry(p_ptr, &ports, port_list) {
773	msg = &p_ptr->publ.phdr;
774	if (msg_orignode(msg) == tipc_own_addr)
775	break;
776	msg_set_prevnode(msg, tipc_own_addr);
777	msg_set_orignode(msg, tipc_own_addr);
778	}
779	spin_unlock_bh(&tipc_port_list_lock);
780	}
781
782
783	/*
784	* port_dispatcher_sigh(): Signal handler for messages destinated
785	* to the tipc_port interface.
786	*/
787
788	static void port_dispatcher_sigh(void *dummy)
789	{
790	struct sk_buff *buf;
791
792	spin_lock_bh(&queue_lock);
793	buf = msg_queue_head;
794	msg_queue_head = NULL;
795	spin_unlock_bh(&queue_lock);
796
797	while (buf) {
798	struct port *p_ptr;
799	struct user_port *up_ptr;
800	struct tipc_portid orig;
801	struct tipc_name_seq dseq;
802	void *usr_handle;
803	int connected;
804	int published;
805	u32 message_type;
806
807	struct sk_buff *next = buf->next;
808	struct tipc_msg *msg = buf_msg(buf);
809	u32 dref = msg_destport(msg);
810
811	message_type = msg_type(msg);
812	if (message_type > TIPC_DIRECT_MSG)
813	goto reject; /* Unsupported message type */
814
815	p_ptr = tipc_port_lock(dref);
816	if (!p_ptr)
817	goto reject; /* Port deleted while msg in queue */
818
819	orig.ref = msg_origport(msg);
820	orig.node = msg_orignode(msg);
821	up_ptr = p_ptr->user_port;
822	usr_handle = up_ptr->usr_handle;
823	connected = p_ptr->publ.connected;
824	published = p_ptr->publ.published;
825
826	if (unlikely(msg_errcode(msg)))
827	goto err;
828
829	switch (message_type) {
830
831	case TIPC_CONN_MSG:{
832	tipc_conn_msg_event cb = up_ptr->conn_msg_cb;
833	u32 peer_port = port_peerport(p_ptr);
834	u32 peer_node = port_peernode(p_ptr);
835
836	tipc_port_unlock(p_ptr);
837	if (unlikely(!cb))
838	goto reject;
839	if (unlikely(!connected)) {
840	if (tipc_connect2port(dref, &orig))
841	goto reject;
842	} else if ((msg_origport(msg) != peer_port) \|\|
843	(msg_orignode(msg) != peer_node))
844	goto reject;
845	if (unlikely(++p_ptr->publ.conn_unacked >=
846	TIPC_FLOW_CONTROL_WIN))
847	tipc_acknowledge(dref,
848	p_ptr->publ.conn_unacked);
849	skb_pull(buf, msg_hdr_sz(msg));
850	cb(usr_handle, dref, &buf, msg_data(msg),
851	msg_data_sz(msg));
852	break;
853	}
854	case TIPC_DIRECT_MSG:{
855	tipc_msg_event cb = up_ptr->msg_cb;
856
857	tipc_port_unlock(p_ptr);
858	if (unlikely(!cb \|\| connected))
859	goto reject;
860	skb_pull(buf, msg_hdr_sz(msg));
861	cb(usr_handle, dref, &buf, msg_data(msg),
862	msg_data_sz(msg), msg_importance(msg),
863	&orig);
864	break;
865	}
866	case TIPC_MCAST_MSG:
867	case TIPC_NAMED_MSG:{
868	tipc_named_msg_event cb = up_ptr->named_msg_cb;
869
870	tipc_port_unlock(p_ptr);
871	if (unlikely(!cb \|\| connected \|\| !published))
872	goto reject;
873	dseq.type = msg_nametype(msg);
874	dseq.lower = msg_nameinst(msg);
875	dseq.upper = (message_type == TIPC_NAMED_MSG)
876	? dseq.lower : msg_nameupper(msg);
877	skb_pull(buf, msg_hdr_sz(msg));
878	cb(usr_handle, dref, &buf, msg_data(msg),
879	msg_data_sz(msg), msg_importance(msg),
880	&orig, &dseq);
881	break;
882	}
883	}
884	if (buf)
885	buf_discard(buf);
886	buf = next;
887	continue;
888	err:
889	switch (message_type) {
890
891	case TIPC_CONN_MSG:{
892	tipc_conn_shutdown_event cb =
893	up_ptr->conn_err_cb;
894	u32 peer_port = port_peerport(p_ptr);
895	u32 peer_node = port_peernode(p_ptr);
896
897	tipc_port_unlock(p_ptr);
898	if (!cb \|\| !connected)
899	break;
900	if ((msg_origport(msg) != peer_port) \|\|
901	(msg_orignode(msg) != peer_node))
902	break;
903	tipc_disconnect(dref);
904	skb_pull(buf, msg_hdr_sz(msg));
905	cb(usr_handle, dref, &buf, msg_data(msg),
906	msg_data_sz(msg), msg_errcode(msg));
907	break;
908	}
909	case TIPC_DIRECT_MSG:{
910	tipc_msg_err_event cb = up_ptr->err_cb;
911
912	tipc_port_unlock(p_ptr);
913	if (!cb \|\| connected)
914	break;
915	skb_pull(buf, msg_hdr_sz(msg));
916	cb(usr_handle, dref, &buf, msg_data(msg),
917	msg_data_sz(msg), msg_errcode(msg), &orig);
918	break;
919	}
920	case TIPC_MCAST_MSG:
921	case TIPC_NAMED_MSG:{
922	tipc_named_msg_err_event cb =
923	up_ptr->named_err_cb;
924
925	tipc_port_unlock(p_ptr);
926	if (!cb \|\| connected)
927	break;
928	dseq.type = msg_nametype(msg);
929	dseq.lower = msg_nameinst(msg);
930	dseq.upper = (message_type == TIPC_NAMED_MSG)
931	? dseq.lower : msg_nameupper(msg);
932	skb_pull(buf, msg_hdr_sz(msg));
933	cb(usr_handle, dref, &buf, msg_data(msg),
934	msg_data_sz(msg), msg_errcode(msg), &dseq);
935	break;
936	}
937	}
938	if (buf)
939	buf_discard(buf);
940	buf = next;
941	continue;
942	reject:
943	tipc_reject_msg(buf, TIPC_ERR_NO_PORT);
944	buf = next;
945	}
946	}
947
948	/*
949	* port_dispatcher(): Dispatcher for messages destinated
950	* to the tipc_port interface. Called with port locked.
951	*/
952
953	static u32 port_dispatcher(struct tipc_port dummy, struct sk_buff buf)
954	{
955	buf->next = NULL;
956	spin_lock_bh(&queue_lock);
957	if (msg_queue_head) {
958	msg_queue_tail->next = buf;
959	msg_queue_tail = buf;
960	} else {
961	msg_queue_tail = msg_queue_head = buf;
962	tipc_k_signal((Handler)port_dispatcher_sigh, 0);
963	}
964	spin_unlock_bh(&queue_lock);
965	return 0;
966	}
967
968	/*
969	* Wake up port after congestion: Called with port locked,
970	*
971	*/
972
973	static void port_wakeup_sh(unsigned long ref)
974	{
975	struct port *p_ptr;
976	struct user_port *up_ptr;
977	tipc_continue_event cb = NULL;
978	void *uh = NULL;
979
980	p_ptr = tipc_port_lock(ref);
981	if (p_ptr) {
982	up_ptr = p_ptr->user_port;
983	if (up_ptr) {
984	cb = up_ptr->continue_event_cb;
985	uh = up_ptr->usr_handle;
986	}
987	tipc_port_unlock(p_ptr);
988	}
989	if (cb)
990	cb(uh, ref);
991	}
992
993
994	static void port_wakeup(struct tipc_port *p_ptr)
995	{
996	tipc_k_signal((Handler)port_wakeup_sh, p_ptr->ref);
997	}
998
999	void tipc_acknowledge(u32 ref, u32 ack)
1000	{
1001	struct port *p_ptr;
1002	struct sk_buff *buf = NULL;
1003
1004	p_ptr = tipc_port_lock(ref);
1005	if (!p_ptr)
1006	return;
1007	if (p_ptr->publ.connected) {
1008	p_ptr->publ.conn_unacked -= ack;
1009	buf = port_build_proto_msg(port_peerport(p_ptr),
1010	port_peernode(p_ptr),
1011	ref,
1012	tipc_own_addr,
1013	CONN_MANAGER,
1014	CONN_ACK,
1015	TIPC_OK,
1016	port_out_seqno(p_ptr),
1017	ack);
1018	}
1019	tipc_port_unlock(p_ptr);
1020	tipc_net_route_msg(buf);
1021	}
1022
1023	/*
1024	* tipc_createport(): user level call. Will add port to
1025	* registry if non-zero user_ref.
1026	*/
1027
1028	int tipc_createport(u32 user_ref,
1029	void *usr_handle,
1030	unsigned int importance,
1031	tipc_msg_err_event error_cb,
1032	tipc_named_msg_err_event named_error_cb,
1033	tipc_conn_shutdown_event conn_error_cb,
1034	tipc_msg_event msg_cb,
1035	tipc_named_msg_event named_msg_cb,
1036	tipc_conn_msg_event conn_msg_cb,
1037	tipc_continue_event continue_event_cb,/* May be zero */
1038	u32 *portref)
1039	{
1040	struct user_port *up_ptr;
1041	struct port *p_ptr;
1042
1043	up_ptr = kmalloc(sizeof(*up_ptr), GFP_ATOMIC);
1044	if (!up_ptr) {
1045	warn("Port creation failed, no memory\n");
1046	return -ENOMEM;
1047	}
1048	p_ptr = (struct port *)tipc_createport_raw(NULL, port_dispatcher,
1049	port_wakeup, importance);
1050	if (!p_ptr) {
1051	kfree(up_ptr);
1052	return -ENOMEM;
1053	}
1054
1055	p_ptr->user_port = up_ptr;
1056	up_ptr->user_ref = user_ref;
1057	up_ptr->usr_handle = usr_handle;
1058	up_ptr->ref = p_ptr->publ.ref;
1059	up_ptr->err_cb = error_cb;
1060	up_ptr->named_err_cb = named_error_cb;
1061	up_ptr->conn_err_cb = conn_error_cb;
1062	up_ptr->msg_cb = msg_cb;
1063	up_ptr->named_msg_cb = named_msg_cb;
1064	up_ptr->conn_msg_cb = conn_msg_cb;
1065	up_ptr->continue_event_cb = continue_event_cb;
1066	INIT_LIST_HEAD(&up_ptr->uport_list);
1067	tipc_reg_add_port(up_ptr);
1068	*portref = p_ptr->publ.ref;
1069	tipc_port_unlock(p_ptr);
1070	return 0;
1071	}
1072
1073	int tipc_ownidentity(u32 ref, struct tipc_portid *id)
1074	{
1075	id->ref = ref;
1076	id->node = tipc_own_addr;
1077	return 0;
1078	}
1079
1080	int tipc_portimportance(u32 ref, unsigned int *importance)
1081	{
1082	struct port *p_ptr;
1083
1084	p_ptr = tipc_port_lock(ref);
1085	if (!p_ptr)
1086	return -EINVAL;
1087	*importance = (unsigned int)msg_importance(&p_ptr->publ.phdr);
1088	tipc_port_unlock(p_ptr);
1089	return 0;
1090	}
1091
1092	int tipc_set_portimportance(u32 ref, unsigned int imp)
1093	{
1094	struct port *p_ptr;
1095
1096	if (imp > TIPC_CRITICAL_IMPORTANCE)
1097	return -EINVAL;
1098
1099	p_ptr = tipc_port_lock(ref);
1100	if (!p_ptr)
1101	return -EINVAL;
1102	msg_set_importance(&p_ptr->publ.phdr, (u32)imp);
1103	tipc_port_unlock(p_ptr);
1104	return 0;
1105	}
1106
1107
1108	int tipc_publish(u32 ref, unsigned int scope, struct tipc_name_seq const *seq)
1109	{
1110	struct port *p_ptr;
1111	struct publication *publ;
1112	u32 key;
1113	int res = -EINVAL;
1114
1115	p_ptr = tipc_port_lock(ref);
1116	if (!p_ptr)
1117	return -EINVAL;
1118
1119	dbg("tipc_publ %u, p_ptr = %x, conn = %x, scope = %x, "
1120	"lower = %u, upper = %u\n",
1121	ref, p_ptr, p_ptr->publ.connected, scope, seq->lower, seq->upper);
1122	if (p_ptr->publ.connected)
1123	goto exit;
1124	if (seq->lower > seq->upper)
1125	goto exit;
1126	if ((scope < TIPC_ZONE_SCOPE) \|\| (scope > TIPC_NODE_SCOPE))
1127	goto exit;
1128	key = ref + p_ptr->pub_count + 1;
1129	if (key == ref) {
1130	res = -EADDRINUSE;
1131	goto exit;
1132	}
1133	publ = tipc_nametbl_publish(seq->type, seq->lower, seq->upper,
1134	scope, p_ptr->publ.ref, key);
1135	if (publ) {
1136	list_add(&publ->pport_list, &p_ptr->publications);
1137	p_ptr->pub_count++;
1138	p_ptr->publ.published = 1;
1139	res = 0;
1140	}
1141	exit:
1142	tipc_port_unlock(p_ptr);
1143	return res;
1144	}
1145
1146	int tipc_withdraw(u32 ref, unsigned int scope, struct tipc_name_seq const *seq)
1147	{
1148	struct port *p_ptr;
1149	struct publication *publ;
1150	struct publication *tpubl;
1151	int res = -EINVAL;
1152
1153	p_ptr = tipc_port_lock(ref);
1154	if (!p_ptr)
1155	return -EINVAL;
1156	if (!seq) {
1157	list_for_each_entry_safe(publ, tpubl,
1158	&p_ptr->publications, pport_list) {
1159	tipc_nametbl_withdraw(publ->type, publ->lower,
1160	publ->ref, publ->key);
1161	}
1162	res = 0;
1163	} else {
1164	list_for_each_entry_safe(publ, tpubl,
1165	&p_ptr->publications, pport_list) {
1166	if (publ->scope != scope)
1167	continue;
1168	if (publ->type != seq->type)
1169	continue;
1170	if (publ->lower != seq->lower)
1171	continue;
1172	if (publ->upper != seq->upper)
1173	break;
1174	tipc_nametbl_withdraw(publ->type, publ->lower,
1175	publ->ref, publ->key);
1176	res = 0;
1177	break;
1178	}
1179	}
1180	if (list_empty(&p_ptr->publications))
1181	p_ptr->publ.published = 0;
1182	tipc_port_unlock(p_ptr);
1183	return res;
1184	}
1185
1186	int tipc_connect2port(u32 ref, struct tipc_portid const *peer)
1187	{
1188	struct port *p_ptr;
1189	struct tipc_msg *msg;
1190	int res = -EINVAL;
1191
1192	p_ptr = tipc_port_lock(ref);
1193	if (!p_ptr)
1194	return -EINVAL;
1195	if (p_ptr->publ.published \|\| p_ptr->publ.connected)
1196	goto exit;
1197	if (!peer->ref)
1198	goto exit;
1199
1200	msg = &p_ptr->publ.phdr;
1201	msg_set_destnode(msg, peer->node);
1202	msg_set_destport(msg, peer->ref);
1203	msg_set_orignode(msg, tipc_own_addr);
1204	msg_set_origport(msg, p_ptr->publ.ref);
1205	msg_set_transp_seqno(msg, 42);
1206	msg_set_type(msg, TIPC_CONN_MSG);
1207	if (!may_route(peer->node))
1208	msg_set_hdr_sz(msg, SHORT_H_SIZE);
1209	else
1210	msg_set_hdr_sz(msg, LONG_H_SIZE);
1211
1212	p_ptr->probing_interval = PROBING_INTERVAL;
1213	p_ptr->probing_state = CONFIRMED;
1214	p_ptr->publ.connected = 1;
1215	k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
1216
1217	tipc_nodesub_subscribe(&p_ptr->subscription,peer->node,
1218	(void *)(unsigned long)ref,
1219	(net_ev_handler)port_handle_node_down);
1220	res = 0;
1221	exit:
1222	tipc_port_unlock(p_ptr);
1223	p_ptr->publ.max_pkt = tipc_link_get_max_pkt(peer->node, ref);
1224	return res;
1225	}
1226
1227	/**
1228	* tipc_disconnect_port - disconnect port from peer
1229	*
1230	* Port must be locked.
1231	*/
1232
1233	int tipc_disconnect_port(struct tipc_port *tp_ptr)
1234	{
1235	int res;
1236
1237	if (tp_ptr->connected) {
1238	tp_ptr->connected = 0;
1239	/* let timer expire on it's own to avoid deadlock! */
1240	tipc_nodesub_unsubscribe(
1241	&((struct port *)tp_ptr)->subscription);
1242	res = 0;
1243	} else {
1244	res = -ENOTCONN;
1245	}
1246	return res;
1247	}
1248
1249	/*
1250	* tipc_disconnect(): Disconnect port form peer.
1251	* This is a node local operation.
1252	*/
1253
1254	int tipc_disconnect(u32 ref)
1255	{
1256	struct port *p_ptr;
1257	int res;
1258
1259	p_ptr = tipc_port_lock(ref);
1260	if (!p_ptr)
1261	return -EINVAL;
1262	res = tipc_disconnect_port((struct tipc_port *)p_ptr);
1263	tipc_port_unlock(p_ptr);
1264	return res;
1265	}
1266
1267	/*
1268	* tipc_shutdown(): Send a SHUTDOWN msg to peer and disconnect
1269	*/
1270	int tipc_shutdown(u32 ref)
1271	{
1272	struct port *p_ptr;
1273	struct sk_buff *buf = NULL;
1274
1275	p_ptr = tipc_port_lock(ref);
1276	if (!p_ptr)
1277	return -EINVAL;
1278
1279	if (p_ptr->publ.connected) {
1280	u32 imp = msg_importance(&p_ptr->publ.phdr);
1281	if (imp < TIPC_CRITICAL_IMPORTANCE)
1282	imp++;
1283	buf = port_build_proto_msg(port_peerport(p_ptr),
1284	port_peernode(p_ptr),
1285	ref,
1286	tipc_own_addr,
1287	imp,
1288	TIPC_CONN_MSG,
1289	TIPC_CONN_SHUTDOWN,
1290	port_out_seqno(p_ptr),
1291	0);
1292	}
1293	tipc_port_unlock(p_ptr);
1294	tipc_net_route_msg(buf);
1295	return tipc_disconnect(ref);
1296	}
1297
1298	int tipc_isconnected(u32 ref, int *isconnected)
1299	{
1300	struct port *p_ptr;
1301
1302	p_ptr = tipc_port_lock(ref);
1303	if (!p_ptr)
1304	return -EINVAL;
1305	*isconnected = p_ptr->publ.connected;
1306	tipc_port_unlock(p_ptr);
1307	return 0;
1308	}
1309
1310	int tipc_peer(u32 ref, struct tipc_portid *peer)
1311	{
1312	struct port *p_ptr;
1313	int res;
1314
1315	p_ptr = tipc_port_lock(ref);
1316	if (!p_ptr)
1317	return -EINVAL;
1318	if (p_ptr->publ.connected) {
1319	peer->ref = port_peerport(p_ptr);
1320	peer->node = port_peernode(p_ptr);
1321	res = 0;
1322	} else
1323	res = -ENOTCONN;
1324	tipc_port_unlock(p_ptr);
1325	return res;
1326	}
1327
1328	int tipc_ref_valid(u32 ref)
1329	{
1330	/* Works irrespective of type */
1331	return !!tipc_ref_deref(ref);
1332	}
1333
1334
1335	/*
1336	* tipc_port_recv_sections(): Concatenate and deliver sectioned
1337	* message for this node.
1338	*/
1339
1340	int tipc_port_recv_sections(struct port *sender, unsigned int num_sect,
1341	struct iovec const *msg_sect)
1342	{
1343	struct sk_buff *buf;
1344	int res;
1345
1346	res = tipc_msg_build(&sender->publ.phdr, msg_sect, num_sect,
1347	MAX_MSG_SIZE, !sender->user_port, &buf);
1348	if (likely(buf))
1349	tipc_port_recv_msg(buf);
1350	return res;
1351	}
1352
1353	/**
1354	* tipc_send - send message sections on connection
1355	*/
1356
1357	int tipc_send(u32 ref, unsigned int num_sect, struct iovec const *msg_sect)
1358	{
1359	struct port *p_ptr;
1360	u32 destnode;
1361	int res;
1362
1363	p_ptr = tipc_port_deref(ref);
1364	if (!p_ptr \|\| !p_ptr->publ.connected)
1365	return -EINVAL;
1366
1367	p_ptr->publ.congested = 1;
1368	if (!tipc_port_congested(p_ptr)) {
1369	destnode = port_peernode(p_ptr);
1370	if (likely(destnode != tipc_own_addr))
1371	res = tipc_link_send_sections_fast(p_ptr, msg_sect, num_sect,
1372	destnode);
1373	else
1374	res = tipc_port_recv_sections(p_ptr, num_sect, msg_sect);
1375
1376	if (likely(res != -ELINKCONG)) {
1377	port_incr_out_seqno(p_ptr);
1378	p_ptr->publ.congested = 0;
1379	p_ptr->sent++;
1380	return res;
1381	}
1382	}
1383	if (port_unreliable(p_ptr)) {
1384	p_ptr->publ.congested = 0;
1385	/* Just calculate msg length and return */
1386	return tipc_msg_calc_data_size(msg_sect, num_sect);
1387	}
1388	return -ELINKCONG;
1389	}
1390
1391	/**
1392	* tipc_send_buf - send message buffer on connection
1393	*/
1394
1395	int tipc_send_buf(u32 ref, struct sk_buff *buf, unsigned int dsz)
1396	{
1397	struct port *p_ptr;
1398	struct tipc_msg *msg;
1399	u32 destnode;
1400	u32 hsz;
1401	u32 sz;
1402	u32 res;
1403
1404	p_ptr = tipc_port_deref(ref);
1405	if (!p_ptr \|\| !p_ptr->publ.connected)
1406	return -EINVAL;
1407
1408	msg = &p_ptr->publ.phdr;
1409	hsz = msg_hdr_sz(msg);
1410	sz = hsz + dsz;
1411	msg_set_size(msg, sz);
1412	if (skb_cow(buf, hsz))
1413	return -ENOMEM;
1414
1415	skb_push(buf, hsz);
1416	skb_copy_to_linear_data(buf, msg, hsz);
1417	destnode = msg_destnode(msg);
1418	p_ptr->publ.congested = 1;
1419	if (!tipc_port_congested(p_ptr)) {
1420	if (likely(destnode != tipc_own_addr))
1421	res = tipc_send_buf_fast(buf, destnode);
1422	else {
1423	tipc_port_recv_msg(buf);
1424	res = sz;
1425	}
1426	if (likely(res != -ELINKCONG)) {
1427	port_incr_out_seqno(p_ptr);
1428	p_ptr->sent++;
1429	p_ptr->publ.congested = 0;
1430	return res;
1431	}
1432	}
1433	if (port_unreliable(p_ptr)) {
1434	p_ptr->publ.congested = 0;
1435	return dsz;
1436	}
1437	return -ELINKCONG;
1438	}
1439
1440	/**
1441	* tipc_forward2name - forward message sections to port name
1442	*/
1443
1444	int tipc_forward2name(u32 ref,
1445	struct tipc_name const *name,
1446	u32 domain,
1447	u32 num_sect,
1448	struct iovec const *msg_sect,
1449	struct tipc_portid const *orig,
1450	unsigned int importance)
1451	{
1452	struct port *p_ptr;
1453	struct tipc_msg *msg;
1454	u32 destnode = domain;
1455	u32 destport;
1456	int res;
1457
1458	p_ptr = tipc_port_deref(ref);
1459	if (!p_ptr \|\| p_ptr->publ.connected)
1460	return -EINVAL;
1461
1462	msg = &p_ptr->publ.phdr;
1463	msg_set_type(msg, TIPC_NAMED_MSG);
1464	msg_set_orignode(msg, orig->node);
1465	msg_set_origport(msg, orig->ref);
1466	msg_set_hdr_sz(msg, LONG_H_SIZE);
1467	msg_set_nametype(msg, name->type);
1468	msg_set_nameinst(msg, name->instance);
1469	msg_set_lookup_scope(msg, tipc_addr_scope(domain));
1470	if (importance <= TIPC_CRITICAL_IMPORTANCE)
1471	msg_set_importance(msg,importance);
1472	destport = tipc_nametbl_translate(name->type, name->instance, &destnode);
1473	msg_set_destnode(msg, destnode);
1474	msg_set_destport(msg, destport);
1475
1476	if (likely(destport \|\| destnode)) {
1477	p_ptr->sent++;
1478	if (likely(destnode == tipc_own_addr))
1479	return tipc_port_recv_sections(p_ptr, num_sect, msg_sect);
1480	res = tipc_link_send_sections_fast(p_ptr, msg_sect, num_sect,
1481	destnode);
1482	if (likely(res != -ELINKCONG))
1483	return res;
1484	if (port_unreliable(p_ptr)) {
1485	/* Just calculate msg length and return */
1486	return tipc_msg_calc_data_size(msg_sect, num_sect);
1487	}
1488	return -ELINKCONG;
1489	}
1490	return tipc_port_reject_sections(p_ptr, msg, msg_sect, num_sect,
1491	TIPC_ERR_NO_NAME);
1492	}
1493
1494	/**
1495	* tipc_send2name - send message sections to port name
1496	*/
1497
1498	int tipc_send2name(u32 ref,
1499	struct tipc_name const *name,
1500	unsigned int domain,
1501	unsigned int num_sect,
1502	struct iovec const *msg_sect)
1503	{
1504	struct tipc_portid orig;
1505
1506	orig.ref = ref;
1507	orig.node = tipc_own_addr;
1508	return tipc_forward2name(ref, name, domain, num_sect, msg_sect, &orig,
1509	TIPC_PORT_IMPORTANCE);
1510	}
1511
1512	/**
1513	* tipc_forward_buf2name - forward message buffer to port name
1514	*/
1515
1516	int tipc_forward_buf2name(u32 ref,
1517	struct tipc_name const *name,
1518	u32 domain,
1519	struct sk_buff *buf,
1520	unsigned int dsz,
1521	struct tipc_portid const *orig,
1522	unsigned int importance)
1523	{
1524	struct port *p_ptr;
1525	struct tipc_msg *msg;
1526	u32 destnode = domain;
1527	u32 destport;
1528	int res;
1529
1530	p_ptr = (struct port *)tipc_ref_deref(ref);
1531	if (!p_ptr \|\| p_ptr->publ.connected)
1532	return -EINVAL;
1533
1534	msg = &p_ptr->publ.phdr;
1535	if (importance <= TIPC_CRITICAL_IMPORTANCE)
1536	msg_set_importance(msg, importance);
1537	msg_set_type(msg, TIPC_NAMED_MSG);
1538	msg_set_orignode(msg, orig->node);
1539	msg_set_origport(msg, orig->ref);
1540	msg_set_nametype(msg, name->type);
1541	msg_set_nameinst(msg, name->instance);
1542	msg_set_lookup_scope(msg, tipc_addr_scope(domain));
1543	msg_set_hdr_sz(msg, LONG_H_SIZE);
1544	msg_set_size(msg, LONG_H_SIZE + dsz);
1545	destport = tipc_nametbl_translate(name->type, name->instance, &destnode);
1546	msg_set_destnode(msg, destnode);
1547	msg_set_destport(msg, destport);
1548	msg_dbg(msg, "forw2name ==> ");
1549	if (skb_cow(buf, LONG_H_SIZE))
1550	return -ENOMEM;
1551	skb_push(buf, LONG_H_SIZE);
1552	skb_copy_to_linear_data(buf, msg, LONG_H_SIZE);
1553	msg_dbg(buf_msg(buf),"PREP:");
1554	if (likely(destport \|\| destnode)) {
1555	p_ptr->sent++;
1556	if (destnode == tipc_own_addr)
1557	return tipc_port_recv_msg(buf);
1558	res = tipc_send_buf_fast(buf, destnode);
1559	if (likely(res != -ELINKCONG))
1560	return res;
1561	if (port_unreliable(p_ptr))
1562	return dsz;
1563	return -ELINKCONG;
1564	}
1565	return tipc_reject_msg(buf, TIPC_ERR_NO_NAME);
1566	}
1567
1568	/**
1569	* tipc_send_buf2name - send message buffer to port name
1570	*/
1571
1572	int tipc_send_buf2name(u32 ref,
1573	struct tipc_name const *dest,
1574	u32 domain,
1575	struct sk_buff *buf,
1576	unsigned int dsz)
1577	{
1578	struct tipc_portid orig;
1579
1580	orig.ref = ref;
1581	orig.node = tipc_own_addr;
1582	return tipc_forward_buf2name(ref, dest, domain, buf, dsz, &orig,
1583	TIPC_PORT_IMPORTANCE);
1584	}
1585
1586	/**
1587	* tipc_forward2port - forward message sections to port identity
1588	*/
1589
1590	int tipc_forward2port(u32 ref,
1591	struct tipc_portid const *dest,
1592	unsigned int num_sect,
1593	struct iovec const *msg_sect,
1594	struct tipc_portid const *orig,
1595	unsigned int importance)
1596	{
1597	struct port *p_ptr;
1598	struct tipc_msg *msg;
1599	int res;
1600
1601	p_ptr = tipc_port_deref(ref);
1602	if (!p_ptr \|\| p_ptr->publ.connected)
1603	return -EINVAL;
1604
1605	msg = &p_ptr->publ.phdr;
1606	msg_set_type(msg, TIPC_DIRECT_MSG);
1607	msg_set_orignode(msg, orig->node);
1608	msg_set_origport(msg, orig->ref);
1609	msg_set_destnode(msg, dest->node);
1610	msg_set_destport(msg, dest->ref);
1611	msg_set_hdr_sz(msg, DIR_MSG_H_SIZE);
1612	if (importance <= TIPC_CRITICAL_IMPORTANCE)
1613	msg_set_importance(msg, importance);
1614	p_ptr->sent++;
1615	if (dest->node == tipc_own_addr)
1616	return tipc_port_recv_sections(p_ptr, num_sect, msg_sect);
1617	res = tipc_link_send_sections_fast(p_ptr, msg_sect, num_sect, dest->node);
1618	if (likely(res != -ELINKCONG))
1619	return res;
1620	if (port_unreliable(p_ptr)) {
1621	/* Just calculate msg length and return */
1622	return tipc_msg_calc_data_size(msg_sect, num_sect);
1623	}
1624	return -ELINKCONG;
1625	}
1626
1627	/**
1628	* tipc_send2port - send message sections to port identity
1629	*/
1630
1631	int tipc_send2port(u32 ref,
1632	struct tipc_portid const *dest,
1633	unsigned int num_sect,
1634	struct iovec const *msg_sect)
1635	{
1636	struct tipc_portid orig;
1637
1638	orig.ref = ref;
1639	orig.node = tipc_own_addr;
1640	return tipc_forward2port(ref, dest, num_sect, msg_sect, &orig,
1641	TIPC_PORT_IMPORTANCE);
1642	}
1643
1644	/**
1645	* tipc_forward_buf2port - forward message buffer to port identity
1646	*/
1647	int tipc_forward_buf2port(u32 ref,
1648	struct tipc_portid const *dest,
1649	struct sk_buff *buf,
1650	unsigned int dsz,
1651	struct tipc_portid const *orig,
1652	unsigned int importance)
1653	{
1654	struct port *p_ptr;
1655	struct tipc_msg *msg;
1656	int res;
1657
1658	p_ptr = (struct port *)tipc_ref_deref(ref);
1659	if (!p_ptr \|\| p_ptr->publ.connected)
1660	return -EINVAL;
1661
1662	msg = &p_ptr->publ.phdr;
1663	msg_set_type(msg, TIPC_DIRECT_MSG);
1664	msg_set_orignode(msg, orig->node);
1665	msg_set_origport(msg, orig->ref);
1666	msg_set_destnode(msg, dest->node);
1667	msg_set_destport(msg, dest->ref);
1668	msg_set_hdr_sz(msg, DIR_MSG_H_SIZE);
1669	if (importance <= TIPC_CRITICAL_IMPORTANCE)
1670	msg_set_importance(msg, importance);
1671	msg_set_size(msg, DIR_MSG_H_SIZE + dsz);
1672	if (skb_cow(buf, DIR_MSG_H_SIZE))
1673	return -ENOMEM;
1674
1675	skb_push(buf, DIR_MSG_H_SIZE);
1676	skb_copy_to_linear_data(buf, msg, DIR_MSG_H_SIZE);
1677	msg_dbg(msg, "buf2port: ");
1678	p_ptr->sent++;
1679	if (dest->node == tipc_own_addr)
1680	return tipc_port_recv_msg(buf);
1681	res = tipc_send_buf_fast(buf, dest->node);
1682	if (likely(res != -ELINKCONG))
1683	return res;
1684	if (port_unreliable(p_ptr))
1685	return dsz;
1686	return -ELINKCONG;
1687	}
1688
1689	/**
1690	* tipc_send_buf2port - send message buffer to port identity
1691	*/
1692
1693	int tipc_send_buf2port(u32 ref,
1694	struct tipc_portid const *dest,
1695	struct sk_buff *buf,
1696	unsigned int dsz)
1697	{
1698	struct tipc_portid orig;
1699
1700	orig.ref = ref;
1701	orig.node = tipc_own_addr;
1702	return tipc_forward_buf2port(ref, dest, buf, dsz, &orig,
1703	TIPC_PORT_IMPORTANCE);
1704	}
1705
1706

Download this file

Branches:
ben-wpan
ben-wpan-stefan
javiroman/ks7010
jz-2.6.34
jz-2.6.34-rc5
jz-2.6.34-rc6
jz-2.6.34-rc7
jz-2.6.35
jz-2.6.36
jz-2.6.37
jz-2.6.38
jz-2.6.39
jz-3.0
jz-3.1
jz-3.11
jz-3.12
jz-3.13
jz-3.15
jz-3.16
jz-3.18-dt
jz-3.2
jz-3.3
jz-3.4
jz-3.5
jz-3.6
jz-3.6-rc2-pwm
jz-3.9
jz-3.9-clk
jz-3.9-rc8
jz47xx
jz47xx-2.6.38
master

Tags:
od-2011-09-04
od-2011-09-18
v2.6.34-rc5
v2.6.34-rc6
v2.6.34-rc7
v3.9

Kernel

Kernel Git Source Tree

Root/net/tipc/port.c

interactive