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
56static struct sk_buff *msg_queue_head = NULL;
57static struct sk_buff *msg_queue_tail = NULL;
58
59DEFINE_SPINLOCK(tipc_port_list_lock);
60static DEFINE_SPINLOCK(queue_lock);
61
62static LIST_HEAD(ports);
63static void port_handle_node_down(unsigned long ref);
64static struct sk_buff* port_build_self_abort_msg(struct port *,u32 err);
65static struct sk_buff* port_build_peer_abort_msg(struct port *,u32 err);
66static void port_timeout(unsigned long ref);
67
68
69static u32 port_peernode(struct port *p_ptr)
70{
71    return msg_destnode(&p_ptr->publ.phdr);
72}
73
74static u32 port_peerport(struct port *p_ptr)
75{
76    return msg_destport(&p_ptr->publ.phdr);
77}
78
79static u32 port_out_seqno(struct port *p_ptr)
80{
81    return msg_transp_seqno(&p_ptr->publ.phdr);
82}
83
84static 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
97int 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
163void 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    }
208exit:
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
219struct 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
262int 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
302struct 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
311void *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
324static int port_unreliable(struct port *p_ptr)
325{
326    return msg_src_droppable(&p_ptr->publ.phdr);
327}
328
329int 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
341int 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
353static int port_unreturnable(struct port *p_ptr)
354{
355    return msg_dest_droppable(&p_ptr->publ.phdr);
356}
357
358int 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
370int 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 */
387static 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
410int 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
476int 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
491static 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
526static 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
539static 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
559static 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
578void 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);
646exit:
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
654static 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
693struct 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
726struct 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
766void 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
788static 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;
888err:
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;
942reject:
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
953static 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
973static 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
994static void port_wakeup(struct tipc_port *p_ptr)
995{
996    tipc_k_signal((Handler)port_wakeup_sh, p_ptr->ref);
997}
998
999void 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
1028int 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
1073int tipc_ownidentity(u32 ref, struct tipc_portid *id)
1074{
1075    id->ref = ref;
1076    id->node = tipc_own_addr;
1077    return 0;
1078}
1079
1080int 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
1092int 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
1108int 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    }
1141exit:
1142    tipc_port_unlock(p_ptr);
1143    return res;
1144}
1145
1146int 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
1186int 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;
1221exit:
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
1233int 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
1254int 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 */
1270int 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
1298int 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
1310int 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
1328int 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
1340int 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
1357int 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
1395int 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
1444int 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
1498int 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
1516int 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
1572int 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
1590int 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
1631int 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 */
1647int 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
1693int 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

Archive Download this file



interactive