tipc: fix skb truesize/datasize ratio control
[muen/linux.git] / net / tipc / msg.c
1 /*
2  * net/tipc/msg.c: TIPC message header routines
3  *
4  * Copyright (c) 2000-2006, 2014-2015, Ericsson AB
5  * Copyright (c) 2005, 2010-2011, 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 <net/sock.h>
38 #include "core.h"
39 #include "msg.h"
40 #include "addr.h"
41 #include "name_table.h"
42
43 #define MAX_FORWARD_SIZE 1024
44 #define BUF_HEADROOM (LL_MAX_HEADER + 48)
45 #define BUF_TAILROOM 16
46
47 static unsigned int align(unsigned int i)
48 {
49         return (i + 3) & ~3u;
50 }
51
52 /**
53  * tipc_buf_acquire - creates a TIPC message buffer
54  * @size: message size (including TIPC header)
55  *
56  * Returns a new buffer with data pointers set to the specified size.
57  *
58  * NOTE: Headroom is reserved to allow prepending of a data link header.
59  *       There may also be unrequested tailroom present at the buffer's end.
60  */
61 struct sk_buff *tipc_buf_acquire(u32 size, gfp_t gfp)
62 {
63         struct sk_buff *skb;
64         unsigned int buf_size = (BUF_HEADROOM + size + 3) & ~3u;
65
66         skb = alloc_skb_fclone(buf_size, gfp);
67         if (skb) {
68                 skb_reserve(skb, BUF_HEADROOM);
69                 skb_put(skb, size);
70                 skb->next = NULL;
71         }
72         return skb;
73 }
74
75 void tipc_msg_init(u32 own_node, struct tipc_msg *m, u32 user, u32 type,
76                    u32 hsize, u32 dnode)
77 {
78         memset(m, 0, hsize);
79         msg_set_version(m);
80         msg_set_user(m, user);
81         msg_set_hdr_sz(m, hsize);
82         msg_set_size(m, hsize);
83         msg_set_prevnode(m, own_node);
84         msg_set_type(m, type);
85         if (hsize > SHORT_H_SIZE) {
86                 msg_set_orignode(m, own_node);
87                 msg_set_destnode(m, dnode);
88         }
89 }
90
91 struct sk_buff *tipc_msg_create(uint user, uint type,
92                                 uint hdr_sz, uint data_sz, u32 dnode,
93                                 u32 onode, u32 dport, u32 oport, int errcode)
94 {
95         struct tipc_msg *msg;
96         struct sk_buff *buf;
97
98         buf = tipc_buf_acquire(hdr_sz + data_sz, GFP_ATOMIC);
99         if (unlikely(!buf))
100                 return NULL;
101
102         msg = buf_msg(buf);
103         tipc_msg_init(onode, msg, user, type, hdr_sz, dnode);
104         msg_set_size(msg, hdr_sz + data_sz);
105         msg_set_origport(msg, oport);
106         msg_set_destport(msg, dport);
107         msg_set_errcode(msg, errcode);
108         if (hdr_sz > SHORT_H_SIZE) {
109                 msg_set_orignode(msg, onode);
110                 msg_set_destnode(msg, dnode);
111         }
112         return buf;
113 }
114
115 /* tipc_buf_append(): Append a buffer to the fragment list of another buffer
116  * @*headbuf: in:  NULL for first frag, otherwise value returned from prev call
117  *            out: set when successful non-complete reassembly, otherwise NULL
118  * @*buf:     in:  the buffer to append. Always defined
119  *            out: head buf after successful complete reassembly, otherwise NULL
120  * Returns 1 when reassembly complete, otherwise 0
121  */
122 int tipc_buf_append(struct sk_buff **headbuf, struct sk_buff **buf)
123 {
124         struct sk_buff *head = *headbuf;
125         struct sk_buff *frag = *buf;
126         struct sk_buff *tail = NULL;
127         struct tipc_msg *msg;
128         u32 fragid;
129         int delta;
130         bool headstolen;
131
132         if (!frag)
133                 goto err;
134
135         msg = buf_msg(frag);
136         fragid = msg_type(msg);
137         frag->next = NULL;
138         skb_pull(frag, msg_hdr_sz(msg));
139
140         if (fragid == FIRST_FRAGMENT) {
141                 if (unlikely(head))
142                         goto err;
143                 if (unlikely(skb_unclone(frag, GFP_ATOMIC)))
144                         goto err;
145                 head = *headbuf = frag;
146                 *buf = NULL;
147                 TIPC_SKB_CB(head)->tail = NULL;
148                 if (skb_is_nonlinear(head)) {
149                         skb_walk_frags(head, tail) {
150                                 TIPC_SKB_CB(head)->tail = tail;
151                         }
152                 } else {
153                         skb_frag_list_init(head);
154                 }
155                 return 0;
156         }
157
158         if (!head)
159                 goto err;
160
161         if (skb_try_coalesce(head, frag, &headstolen, &delta)) {
162                 kfree_skb_partial(frag, headstolen);
163         } else {
164                 tail = TIPC_SKB_CB(head)->tail;
165                 if (!skb_has_frag_list(head))
166                         skb_shinfo(head)->frag_list = frag;
167                 else
168                         tail->next = frag;
169                 head->truesize += frag->truesize;
170                 head->data_len += frag->len;
171                 head->len += frag->len;
172                 TIPC_SKB_CB(head)->tail = frag;
173         }
174
175         if (fragid == LAST_FRAGMENT) {
176                 TIPC_SKB_CB(head)->validated = false;
177                 if (unlikely(!tipc_msg_validate(&head)))
178                         goto err;
179                 *buf = head;
180                 TIPC_SKB_CB(head)->tail = NULL;
181                 *headbuf = NULL;
182                 return 1;
183         }
184         *buf = NULL;
185         return 0;
186 err:
187         kfree_skb(*buf);
188         kfree_skb(*headbuf);
189         *buf = *headbuf = NULL;
190         return 0;
191 }
192
193 /* tipc_msg_validate - validate basic format of received message
194  *
195  * This routine ensures a TIPC message has an acceptable header, and at least
196  * as much data as the header indicates it should.  The routine also ensures
197  * that the entire message header is stored in the main fragment of the message
198  * buffer, to simplify future access to message header fields.
199  *
200  * Note: Having extra info present in the message header or data areas is OK.
201  * TIPC will ignore the excess, under the assumption that it is optional info
202  * introduced by a later release of the protocol.
203  */
204 bool tipc_msg_validate(struct sk_buff **_skb)
205 {
206         struct sk_buff *skb = *_skb;
207         struct tipc_msg *hdr;
208         int msz, hsz;
209
210         /* Ensure that flow control ratio condition is satisfied */
211         if (unlikely(skb->truesize / buf_roundup_len(skb) >= 4)) {
212                 skb = skb_copy_expand(skb, BUF_HEADROOM, 0, GFP_ATOMIC);
213                 if (!skb)
214                         return false;
215                 kfree_skb(*_skb);
216                 *_skb = skb;
217         }
218
219         if (unlikely(TIPC_SKB_CB(skb)->validated))
220                 return true;
221         if (unlikely(!pskb_may_pull(skb, MIN_H_SIZE)))
222                 return false;
223
224         hsz = msg_hdr_sz(buf_msg(skb));
225         if (unlikely(hsz < MIN_H_SIZE) || (hsz > MAX_H_SIZE))
226                 return false;
227         if (unlikely(!pskb_may_pull(skb, hsz)))
228                 return false;
229
230         hdr = buf_msg(skb);
231         if (unlikely(msg_version(hdr) != TIPC_VERSION))
232                 return false;
233
234         msz = msg_size(hdr);
235         if (unlikely(msz < hsz))
236                 return false;
237         if (unlikely((msz - hsz) > TIPC_MAX_USER_MSG_SIZE))
238                 return false;
239         if (unlikely(skb->len < msz))
240                 return false;
241
242         TIPC_SKB_CB(skb)->validated = true;
243         return true;
244 }
245
246 /**
247  * tipc_msg_build - create buffer chain containing specified header and data
248  * @mhdr: Message header, to be prepended to data
249  * @m: User message
250  * @dsz: Total length of user data
251  * @pktmax: Max packet size that can be used
252  * @list: Buffer or chain of buffers to be returned to caller
253  *
254  * Note that the recursive call we are making here is safe, since it can
255  * logically go only one further level down.
256  *
257  * Returns message data size or errno: -ENOMEM, -EFAULT
258  */
259 int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m, int offset,
260                    int dsz, int pktmax, struct sk_buff_head *list)
261 {
262         int mhsz = msg_hdr_sz(mhdr);
263         struct tipc_msg pkthdr;
264         int msz = mhsz + dsz;
265         int pktrem = pktmax;
266         struct sk_buff *skb;
267         int drem = dsz;
268         int pktno = 1;
269         char *pktpos;
270         int pktsz;
271         int rc;
272
273         msg_set_size(mhdr, msz);
274
275         /* No fragmentation needed? */
276         if (likely(msz <= pktmax)) {
277                 skb = tipc_buf_acquire(msz, GFP_KERNEL);
278
279                 /* Fall back to smaller MTU if node local message */
280                 if (unlikely(!skb)) {
281                         if (pktmax != MAX_MSG_SIZE)
282                                 return -ENOMEM;
283                         rc = tipc_msg_build(mhdr, m, offset, dsz, FB_MTU, list);
284                         if (rc != dsz)
285                                 return rc;
286                         if (tipc_msg_assemble(list))
287                                 return dsz;
288                         return -ENOMEM;
289                 }
290                 skb_orphan(skb);
291                 __skb_queue_tail(list, skb);
292                 skb_copy_to_linear_data(skb, mhdr, mhsz);
293                 pktpos = skb->data + mhsz;
294                 if (copy_from_iter_full(pktpos, dsz, &m->msg_iter))
295                         return dsz;
296                 rc = -EFAULT;
297                 goto error;
298         }
299
300         /* Prepare reusable fragment header */
301         tipc_msg_init(msg_prevnode(mhdr), &pkthdr, MSG_FRAGMENTER,
302                       FIRST_FRAGMENT, INT_H_SIZE, msg_destnode(mhdr));
303         msg_set_size(&pkthdr, pktmax);
304         msg_set_fragm_no(&pkthdr, pktno);
305         msg_set_importance(&pkthdr, msg_importance(mhdr));
306
307         /* Prepare first fragment */
308         skb = tipc_buf_acquire(pktmax, GFP_KERNEL);
309         if (!skb)
310                 return -ENOMEM;
311         skb_orphan(skb);
312         __skb_queue_tail(list, skb);
313         pktpos = skb->data;
314         skb_copy_to_linear_data(skb, &pkthdr, INT_H_SIZE);
315         pktpos += INT_H_SIZE;
316         pktrem -= INT_H_SIZE;
317         skb_copy_to_linear_data_offset(skb, INT_H_SIZE, mhdr, mhsz);
318         pktpos += mhsz;
319         pktrem -= mhsz;
320
321         do {
322                 if (drem < pktrem)
323                         pktrem = drem;
324
325                 if (!copy_from_iter_full(pktpos, pktrem, &m->msg_iter)) {
326                         rc = -EFAULT;
327                         goto error;
328                 }
329                 drem -= pktrem;
330
331                 if (!drem)
332                         break;
333
334                 /* Prepare new fragment: */
335                 if (drem < (pktmax - INT_H_SIZE))
336                         pktsz = drem + INT_H_SIZE;
337                 else
338                         pktsz = pktmax;
339                 skb = tipc_buf_acquire(pktsz, GFP_KERNEL);
340                 if (!skb) {
341                         rc = -ENOMEM;
342                         goto error;
343                 }
344                 skb_orphan(skb);
345                 __skb_queue_tail(list, skb);
346                 msg_set_type(&pkthdr, FRAGMENT);
347                 msg_set_size(&pkthdr, pktsz);
348                 msg_set_fragm_no(&pkthdr, ++pktno);
349                 skb_copy_to_linear_data(skb, &pkthdr, INT_H_SIZE);
350                 pktpos = skb->data + INT_H_SIZE;
351                 pktrem = pktsz - INT_H_SIZE;
352
353         } while (1);
354         msg_set_type(buf_msg(skb), LAST_FRAGMENT);
355         return dsz;
356 error:
357         __skb_queue_purge(list);
358         __skb_queue_head_init(list);
359         return rc;
360 }
361
362 /**
363  * tipc_msg_bundle(): Append contents of a buffer to tail of an existing one
364  * @skb: the buffer to append to ("bundle")
365  * @msg:  message to be appended
366  * @mtu:  max allowable size for the bundle buffer
367  * Consumes buffer if successful
368  * Returns true if bundling could be performed, otherwise false
369  */
370 bool tipc_msg_bundle(struct sk_buff *skb, struct tipc_msg *msg, u32 mtu)
371 {
372         struct tipc_msg *bmsg;
373         unsigned int bsz;
374         unsigned int msz = msg_size(msg);
375         u32 start, pad;
376         u32 max = mtu - INT_H_SIZE;
377
378         if (likely(msg_user(msg) == MSG_FRAGMENTER))
379                 return false;
380         if (!skb)
381                 return false;
382         bmsg = buf_msg(skb);
383         bsz = msg_size(bmsg);
384         start = align(bsz);
385         pad = start - bsz;
386
387         if (unlikely(msg_user(msg) == TUNNEL_PROTOCOL))
388                 return false;
389         if (unlikely(msg_user(msg) == BCAST_PROTOCOL))
390                 return false;
391         if (unlikely(msg_user(bmsg) != MSG_BUNDLER))
392                 return false;
393         if (unlikely(skb_tailroom(skb) < (pad + msz)))
394                 return false;
395         if (unlikely(max < (start + msz)))
396                 return false;
397         if ((msg_importance(msg) < TIPC_SYSTEM_IMPORTANCE) &&
398             (msg_importance(bmsg) == TIPC_SYSTEM_IMPORTANCE))
399                 return false;
400
401         skb_put(skb, pad + msz);
402         skb_copy_to_linear_data_offset(skb, start, msg, msz);
403         msg_set_size(bmsg, start + msz);
404         msg_set_msgcnt(bmsg, msg_msgcnt(bmsg) + 1);
405         return true;
406 }
407
408 /**
409  *  tipc_msg_extract(): extract bundled inner packet from buffer
410  *  @skb: buffer to be extracted from.
411  *  @iskb: extracted inner buffer, to be returned
412  *  @pos: position in outer message of msg to be extracted.
413  *        Returns position of next msg
414  *  Consumes outer buffer when last packet extracted
415  *  Returns true when when there is an extracted buffer, otherwise false
416  */
417 bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos)
418 {
419         struct tipc_msg *msg;
420         int imsz, offset;
421
422         *iskb = NULL;
423         if (unlikely(skb_linearize(skb)))
424                 goto none;
425
426         msg = buf_msg(skb);
427         offset = msg_hdr_sz(msg) + *pos;
428         if (unlikely(offset > (msg_size(msg) - MIN_H_SIZE)))
429                 goto none;
430
431         *iskb = skb_clone(skb, GFP_ATOMIC);
432         if (unlikely(!*iskb))
433                 goto none;
434         skb_pull(*iskb, offset);
435         imsz = msg_size(buf_msg(*iskb));
436         skb_trim(*iskb, imsz);
437         if (unlikely(!tipc_msg_validate(iskb)))
438                 goto none;
439         *pos += align(imsz);
440         return true;
441 none:
442         kfree_skb(skb);
443         kfree_skb(*iskb);
444         *iskb = NULL;
445         return false;
446 }
447
448 /**
449  * tipc_msg_make_bundle(): Create bundle buf and append message to its tail
450  * @list: the buffer chain, where head is the buffer to replace/append
451  * @skb: buffer to be created, appended to and returned in case of success
452  * @msg: message to be appended
453  * @mtu: max allowable size for the bundle buffer, inclusive header
454  * @dnode: destination node for message. (Not always present in header)
455  * Returns true if success, otherwise false
456  */
457 bool tipc_msg_make_bundle(struct sk_buff **skb,  struct tipc_msg *msg,
458                           u32 mtu, u32 dnode)
459 {
460         struct sk_buff *_skb;
461         struct tipc_msg *bmsg;
462         u32 msz = msg_size(msg);
463         u32 max = mtu - INT_H_SIZE;
464
465         if (msg_user(msg) == MSG_FRAGMENTER)
466                 return false;
467         if (msg_user(msg) == TUNNEL_PROTOCOL)
468                 return false;
469         if (msg_user(msg) == BCAST_PROTOCOL)
470                 return false;
471         if (msz > (max / 2))
472                 return false;
473
474         _skb = tipc_buf_acquire(max, GFP_ATOMIC);
475         if (!_skb)
476                 return false;
477
478         skb_trim(_skb, INT_H_SIZE);
479         bmsg = buf_msg(_skb);
480         tipc_msg_init(msg_prevnode(msg), bmsg, MSG_BUNDLER, 0,
481                       INT_H_SIZE, dnode);
482         if (msg_isdata(msg))
483                 msg_set_importance(bmsg, TIPC_CRITICAL_IMPORTANCE);
484         else
485                 msg_set_importance(bmsg, TIPC_SYSTEM_IMPORTANCE);
486         msg_set_seqno(bmsg, msg_seqno(msg));
487         msg_set_ack(bmsg, msg_ack(msg));
488         msg_set_bcast_ack(bmsg, msg_bcast_ack(msg));
489         tipc_msg_bundle(_skb, msg, mtu);
490         *skb = _skb;
491         return true;
492 }
493
494 /**
495  * tipc_msg_reverse(): swap source and destination addresses and add error code
496  * @own_node: originating node id for reversed message
497  * @skb:  buffer containing message to be reversed; may be replaced.
498  * @err:  error code to be set in message, if any
499  * Consumes buffer at failure
500  * Returns true if success, otherwise false
501  */
502 bool tipc_msg_reverse(u32 own_node,  struct sk_buff **skb, int err)
503 {
504         struct sk_buff *_skb = *skb;
505         struct tipc_msg *hdr;
506         struct tipc_msg ohdr;
507         int dlen;
508
509         if (skb_linearize(_skb))
510                 goto exit;
511         hdr = buf_msg(_skb);
512         dlen = min_t(uint, msg_data_sz(hdr), MAX_FORWARD_SIZE);
513         if (msg_dest_droppable(hdr))
514                 goto exit;
515         if (msg_errcode(hdr))
516                 goto exit;
517
518         /* Take a copy of original header before altering message */
519         memcpy(&ohdr, hdr, msg_hdr_sz(hdr));
520
521         /* Never return SHORT header; expand by replacing buffer if necessary */
522         if (msg_short(hdr)) {
523                 *skb = tipc_buf_acquire(BASIC_H_SIZE + dlen, GFP_ATOMIC);
524                 if (!*skb)
525                         goto exit;
526                 memcpy((*skb)->data + BASIC_H_SIZE, msg_data(hdr), dlen);
527                 kfree_skb(_skb);
528                 _skb = *skb;
529                 hdr = buf_msg(_skb);
530                 memcpy(hdr, &ohdr, BASIC_H_SIZE);
531                 msg_set_hdr_sz(hdr, BASIC_H_SIZE);
532         }
533
534         if (skb_cloned(_skb) &&
535             pskb_expand_head(_skb, BUF_HEADROOM, BUF_TAILROOM, GFP_ATOMIC))
536                 goto exit;
537
538         /* reassign after skb header modifications */
539         hdr = buf_msg(_skb);
540         /* Now reverse the concerned fields */
541         msg_set_errcode(hdr, err);
542         msg_set_non_seq(hdr, 0);
543         msg_set_origport(hdr, msg_destport(&ohdr));
544         msg_set_destport(hdr, msg_origport(&ohdr));
545         msg_set_destnode(hdr, msg_prevnode(&ohdr));
546         msg_set_prevnode(hdr, own_node);
547         msg_set_orignode(hdr, own_node);
548         msg_set_size(hdr, msg_hdr_sz(hdr) + dlen);
549         skb_trim(_skb, msg_size(hdr));
550         skb_orphan(_skb);
551         return true;
552 exit:
553         kfree_skb(_skb);
554         *skb = NULL;
555         return false;
556 }
557
558 /**
559  * tipc_msg_lookup_dest(): try to find new destination for named message
560  * @skb: the buffer containing the message.
561  * @err: error code to be used by caller if lookup fails
562  * Does not consume buffer
563  * Returns true if a destination is found, false otherwise
564  */
565 bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, int *err)
566 {
567         struct tipc_msg *msg = buf_msg(skb);
568         u32 dport, dnode;
569         u32 onode = tipc_own_addr(net);
570
571         if (!msg_isdata(msg))
572                 return false;
573         if (!msg_named(msg))
574                 return false;
575         if (msg_errcode(msg))
576                 return false;
577         *err = TIPC_ERR_NO_NAME;
578         if (skb_linearize(skb))
579                 return false;
580         msg = buf_msg(skb);
581         if (msg_reroute_cnt(msg))
582                 return false;
583         dnode = addr_domain(net, msg_lookup_scope(msg));
584         dport = tipc_nametbl_translate(net, msg_nametype(msg),
585                                        msg_nameinst(msg), &dnode);
586         if (!dport)
587                 return false;
588         msg_incr_reroute_cnt(msg);
589         if (dnode != onode)
590                 msg_set_prevnode(msg, onode);
591         msg_set_destnode(msg, dnode);
592         msg_set_destport(msg, dport);
593         *err = TIPC_OK;
594
595         if (!skb_cloned(skb))
596                 return true;
597
598         /* Unclone buffer in case it was bundled */
599         if (pskb_expand_head(skb, BUF_HEADROOM, BUF_TAILROOM, GFP_ATOMIC))
600                 return false;
601
602         return true;
603 }
604
605 /* tipc_msg_assemble() - assemble chain of fragments into one message
606  */
607 bool tipc_msg_assemble(struct sk_buff_head *list)
608 {
609         struct sk_buff *skb, *tmp = NULL;
610
611         if (skb_queue_len(list) == 1)
612                 return true;
613
614         while ((skb = __skb_dequeue(list))) {
615                 skb->next = NULL;
616                 if (tipc_buf_append(&tmp, &skb)) {
617                         __skb_queue_tail(list, skb);
618                         return true;
619                 }
620                 if (!tmp)
621                         break;
622         }
623         __skb_queue_purge(list);
624         __skb_queue_head_init(list);
625         pr_warn("Failed do assemble buffer\n");
626         return false;
627 }
628
629 /* tipc_msg_reassemble() - clone a buffer chain of fragments and
630  *                         reassemble the clones into one message
631  */
632 bool tipc_msg_reassemble(struct sk_buff_head *list, struct sk_buff_head *rcvq)
633 {
634         struct sk_buff *skb, *_skb;
635         struct sk_buff *frag = NULL;
636         struct sk_buff *head = NULL;
637         int hdr_len;
638
639         /* Copy header if single buffer */
640         if (skb_queue_len(list) == 1) {
641                 skb = skb_peek(list);
642                 hdr_len = skb_headroom(skb) + msg_hdr_sz(buf_msg(skb));
643                 _skb = __pskb_copy(skb, hdr_len, GFP_ATOMIC);
644                 if (!_skb)
645                         return false;
646                 __skb_queue_tail(rcvq, _skb);
647                 return true;
648         }
649
650         /* Clone all fragments and reassemble */
651         skb_queue_walk(list, skb) {
652                 frag = skb_clone(skb, GFP_ATOMIC);
653                 if (!frag)
654                         goto error;
655                 frag->next = NULL;
656                 if (tipc_buf_append(&head, &frag))
657                         break;
658                 if (!head)
659                         goto error;
660         }
661         __skb_queue_tail(rcvq, frag);
662         return true;
663 error:
664         pr_warn("Failed do clone local mcast rcv buffer\n");
665         kfree_skb(head);
666         return false;
667 }
668
669 bool tipc_msg_pskb_copy(u32 dst, struct sk_buff_head *msg,
670                         struct sk_buff_head *cpy)
671 {
672         struct sk_buff *skb, *_skb;
673
674         skb_queue_walk(msg, skb) {
675                 _skb = pskb_copy(skb, GFP_ATOMIC);
676                 if (!_skb) {
677                         __skb_queue_purge(cpy);
678                         return false;
679                 }
680                 msg_set_destnode(buf_msg(_skb), dst);
681                 __skb_queue_tail(cpy, _skb);
682         }
683         return true;
684 }
685
686 /* tipc_skb_queue_sorted(); sort pkt into list according to sequence number
687  * @list: list to be appended to
688  * @seqno: sequence number of buffer to add
689  * @skb: buffer to add
690  */
691 void __tipc_skb_queue_sorted(struct sk_buff_head *list, u16 seqno,
692                              struct sk_buff *skb)
693 {
694         struct sk_buff *_skb, *tmp;
695
696         if (skb_queue_empty(list) || less(seqno, buf_seqno(skb_peek(list)))) {
697                 __skb_queue_head(list, skb);
698                 return;
699         }
700
701         if (more(seqno, buf_seqno(skb_peek_tail(list)))) {
702                 __skb_queue_tail(list, skb);
703                 return;
704         }
705
706         skb_queue_walk_safe(list, _skb, tmp) {
707                 if (more(seqno, buf_seqno(_skb)))
708                         continue;
709                 if (seqno == buf_seqno(_skb))
710                         break;
711                 __skb_queue_before(list, _skb, skb);
712                 return;
713         }
714         kfree_skb(skb);
715 }
716
717 void tipc_skb_reject(struct net *net, int err, struct sk_buff *skb,
718                      struct sk_buff_head *xmitq)
719 {
720         if (tipc_msg_reverse(tipc_own_addr(net), &skb, err))
721                 __skb_queue_tail(xmitq, skb);
722 }