aboutsummaryrefslogtreecommitdiffstats
path: root/src/vnet/dpo/replicate_dpo.c
blob: 69e367aa3686412690151214999bd50ffadb2ac8 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
/*
 * Copyright (c) 2018 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef included_time_range_h
#define included_time_range_h

#include <vppinfra/format.h>
#include <vppinfra/time.h>

typedef enum
{
  CLIB_TIMEBASE_DAYLIGHT_NONE = 0,
  CLIB_TIMEBASE_DAYLIGHT_USA,
} clib_timebase_daylight_time_t;

typedef struct
{
  /* provides f64 seconds since clib_time_init was called */
  clib_time_t clib_time;
  /*
   * time in f64 seconds since Thursday 1 Jan 1970 00:00:00 UTC
   * when clib_time_init was called
   */
  f64 time_zero;
  f64 timezone_offset;
  f64 summer_offset;
  clib_timebase_daylight_time_t daylight_time_type;
  f64 cached_year_start;
  f64 cached_year_end;
  f64 cached_summer_start;
  f64 cached_summer_end;
} clib_timebase_t;

typedef struct
{
  u32 year, month, day, hour, minute, second, nanosecond;
  /* 0 => Thursday */
  u32 day_name_index;
  f64 fractional_seconds;
} clib_timebase_component_t;

typedef struct
{
  f64 start, end;
} clib_timebase_range_t;

void clib_timebase_init (clib_timebase_t * tb, i32 timezone_offset_in_hours,
			 clib_timebase_daylight_time_t daylight_type);

void clib_timebase_time_to_components (f64 now,
				       clib_timebase_component_t * cp);

f64 clib_timebase_components_to_time (clib_timebase_component_t * cp);

f64 clib_timebase_find_sunday_midnight (f64 start_time);
f64 clib_timebase_offset_from_sunday (u8 * day);
f64 clib_timebase_summer_offset (clib_timebase_t * tb, f64 now);

unformat_function_t unformat_clib_timebase_range_hms;
unformat_function_t unformat_clib_timebase_range_vector;

format_function_t format_clib_timebase_time;

static inline f64 clib_timebase_summer_offset_fastpath
  (clib_timebase_t * tb, f64 now)
{
  if (PREDICT_TRUE
      (now >= tb->cached_year_start && now <= tb->cached_year_end))
    {
      if (now >= tb->cached_summer_start && now <= tb->cached_summer_end)
	return tb->summer_offset;
      else
	return 0.0;
    }
  else
    return clib_timebase_summer_offset (tb, now);
}

static inline f64
clib_timebase_now (clib_timebase_t * tb)
{
  f64 now;

  now = tb->time_zero + clib_time_now (&tb->clib_time);
  now += tb->timezone_offset;
  now += clib_timebase_summer_offset_fastpath (tb, now);

  return now;
}

static inline int
clib_timebase_is_leap_year (u32 year)
{
  int rv;

  if (PREDICT_TRUE ((year % 4) != 0))
    return 0;

  rv = 0;

  if ((year % 4) == 0)
    rv = 1;
  if ((year % 100) == 0)
    rv = 0;
  if ((year % 400) == 0)
    rv = 1;
  return rv;
}

#endif /* included_time_range_h */


/*
 * fd.io coding-style-patch-verification: ON
 *
 * Local Variables:
 * eval: (c-set-style "gnu")
 * End:
 */
href='#n757'>757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920
/*
 * Copyright (c) 2016 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <vnet/ip/lookup.h>
#include <vnet/dpo/replicate_dpo.h>
#include <vnet/dpo/drop_dpo.h>
#include <vnet/dpo/receive_dpo.h>
#include <vnet/adj/adj.h>
#include <vnet/mpls/mpls_types.h>

/**
 * the logger
 */
vlib_log_class_t replicate_logger;

#define REP_DBG(_rep, _fmt, _args...)                                   \
{                                                                       \
    vlib_log_debug(replicate_logger,                                    \
                   "rep:[%U]:" _fmt,                                    \
                   format_replicate,                                    \
                   replicate_get_index(_rep),                           \
                   REPLICATE_FORMAT_NONE,                               \
                   ##_args);                                            \
}

#define foreach_replicate_dpo_error                       \
_(BUFFER_ALLOCATION_FAILURE, "Buffer Allocation Failure")

typedef enum {
#define _(sym,str) REPLICATE_DPO_ERROR_##sym,
  foreach_replicate_dpo_error
#undef _
  REPLICATE_DPO_N_ERROR,
} replicate_dpo_error_t;

static char * replicate_dpo_error_strings[] = {
#define _(sym,string) string,
  foreach_replicate_dpo_error
#undef _
};

/**
 * Pool of all DPOs. It's not static so the DP can have fast access
 */
replicate_t *replicate_pool;

/**
 * The one instance of replicate main
 */
replicate_main_t replicate_main = {
    .repm_counters = {
        .name = "mroutes",
        .stat_segment_name = "/net/mroute",
    },
};

static inline index_t
replicate_get_index (const replicate_t *rep)
{
    return (rep - replicate_pool);
}

static inline dpo_id_t*
replicate_get_buckets (replicate_t *rep)
{
    if (REP_HAS_INLINE_BUCKETS(rep))
    {
        return (rep->rep_buckets_inline);
    }
    else
    {
        return (rep->rep_buckets);
    }
}

static replicate_t *
replicate_alloc_i (void)
{
    replicate_t *rep;

    pool_get_aligned(replicate_pool, rep, CLIB_CACHE_LINE_BYTES);
    clib_memset(rep, 0, sizeof(*rep));

    vlib_validate_combined_counter(&(replicate_main.repm_counters),
                                   replicate_get_index(rep));
    vlib_zero_combined_counter(&(replicate_main.repm_counters),
                               replicate_get_index(rep));

    return (rep);
}

static u8*
format_replicate_flags (u8 *s, va_list *args)
{
    int flags = va_arg (*args, int);

    if (flags == REPLICATE_FLAGS_NONE)
    {
        s = format (s, "none");
    }
    else if (flags & REPLICATE_FLAGS_HAS_LOCAL)
    {
        s = format (s, "has-local ");
    }

    return (s);
}

static u8*
replicate_format (index_t repi,
                  replicate_format_flags_t flags,
                  u32 indent,
                  u8 *s)
{
    vlib_counter_t to;
    replicate_t *rep;
    dpo_id_t *buckets;
    u32 i;

    repi &= ~MPLS_IS_REPLICATE;
    rep = replicate_get(repi);
    vlib_get_combined_counter(&(replicate_main.repm_counters), repi, &to);
    buckets = replicate_get_buckets(rep);

    s = format(s, "%U: ", format_dpo_type, DPO_REPLICATE);
    s = format(s, "[index:%d buckets:%d ", repi, rep->rep_n_buckets);
    s = format(s, "flags:[%U] ", format_replicate_flags, rep->rep_flags);
    s = format(s, "to:[%Ld:%Ld]]", to.packets, to.bytes);

    for (i = 0; i < rep->rep_n_buckets; i++)
    {
        s = format(s, "\n%U", format_white_space, indent+2);
        s = format(s, "[%d]", i);
        s = format(s, " %U", format_dpo_id, &buckets[i], indent+6);
    }
    return (s);
}

u8*
format_replicate (u8 * s, va_list * args)
{
    index_t repi = va_arg(*args, index_t);
    replicate_format_flags_t flags = va_arg(*args, replicate_format_flags_t);

    return (replicate_format(repi, flags, 0, s));
}
static u8*
format_replicate_dpo (u8 * s, va_list * args)
{
    index_t repi = va_arg(*args, index_t);
    u32 indent = va_arg(*args, u32);

    return (replicate_format(repi, REPLICATE_FORMAT_DETAIL, indent, s));
}


static replicate_t *
replicate_create_i (u32 num_buckets,
                    dpo_proto_t rep_proto)
{
    replicate_t *rep;

    rep = replicate_alloc_i();
    rep->rep_n_buckets = num_buckets;
    rep->rep_proto = rep_proto;

    if (!REP_HAS_INLINE_BUCKETS(rep))
    {
        vec_validate_aligned(rep->rep_buckets,
                             rep->rep_n_buckets - 1,
                             CLIB_CACHE_LINE_BYTES);
    }

    REP_DBG(rep, "create");

    return (rep);
}

index_t
replicate_create (u32 n_buckets,
                  dpo_proto_t rep_proto)
{
    return (replicate_get_index(replicate_create_i(n_buckets, rep_proto)));
}

static inline void
replicate_set_bucket_i (replicate_t *rep,
                        u32 bucket,
                        dpo_id_t *buckets,
                        const dpo_id_t *next)
{
    if (dpo_is_receive(&buckets[bucket]))
    {
        rep->rep_flags &= ~REPLICATE_FLAGS_HAS_LOCAL;
    }
    if (dpo_is_receive(next))
    {
        rep->rep_flags |= REPLICATE_FLAGS_HAS_LOCAL;
    }
    dpo_stack(DPO_REPLICATE, rep->rep_proto, &buckets[bucket], next);
}

void
replicate_set_bucket (index_t repi,
                      u32 bucket,
                      const dpo_id_t *next)
{
    replicate_t *rep;
    dpo_id_t *buckets;

    repi &= ~MPLS_IS_REPLICATE;
    rep = replicate_get(repi);
    buckets = replicate_get_buckets(rep);

    ASSERT(bucket < rep->rep_n_buckets);

    replicate_set_bucket_i(rep, bucket, buckets, next);
}

int
replicate_is_drop (const dpo_id_t *dpo)
{
    replicate_t *rep;
    index_t repi;

    if (DPO_REPLICATE != dpo->dpoi_type)
        return (0);

    repi = dpo->dpoi_index & ~MPLS_IS_REPLICATE;
    rep = replicate_get(repi);

    if (1 == rep->rep_n_buckets)
    {
        return (dpo_is_drop(replicate_get_bucket_i(rep, 0)));
    }
    return (0);
}

const dpo_id_t *
replicate_get_bucket (index_t repi,
                      u32 bucket)
{
    replicate_t *rep;

    repi &= ~MPLS_IS_REPLICATE;
    rep = replicate_get(repi);

    return (replicate_get_bucket_i(rep, bucket));
}


static load_balance_path_t *
replicate_multipath_next_hop_fixup (load_balance_path_t *nhs,
                                    dpo_proto_t drop_proto)
{
    if (0 == vec_len(nhs))
    {
        load_balance_path_t *nh;

        /*
         * we need something for the replicate. so use the drop
         */
        vec_add2(nhs, nh, 1);

        nh->path_weight = 1;
        dpo_copy(&nh->path_dpo, drop_dpo_get(drop_proto));
    }

    return (nhs);
}

/*
 * Fill in adjacencies in block based on corresponding
 * next hop adjacencies.
 */
static void
replicate_fill_buckets (replicate_t *rep,
                        load_balance_path_t *nhs,
                        dpo_id_t *buckets,
                        u32 n_buckets)
{
    load_balance_path_t * nh;
    u16 bucket;

    bucket = 0;

    /*
     * the next-hops have normalised weights. that means their sum is the number
     * of buckets we need to fill.
     */
    vec_foreach (nh, nhs)
    {
        ASSERT(bucket < n_buckets);
        replicate_set_bucket_i(rep, bucket++, buckets, &nh->path_dpo);
    }
}

static inline void
replicate_set_n_buckets (replicate_t *rep,
                         u32 n_buckets)
{
    rep->rep_n_buckets = n_buckets;
}

void
replicate_multipath_update (const dpo_id_t *dpo,
                            load_balance_path_t * next_hops)
{
    load_balance_path_t * nh, * nhs;
    dpo_id_t *tmp_dpo;
    u32 ii, n_buckets;
    replicate_t *rep;
    index_t repi;

    ASSERT(DPO_REPLICATE == dpo->dpoi_type);
    repi = dpo->dpoi_index & ~MPLS_IS_REPLICATE;
    rep = replicate_get(repi);
    nhs = replicate_multipath_next_hop_fixup(next_hops,
                                             rep->rep_proto);
    n_buckets = vec_len(nhs);

    if (0 == rep->rep_n_buckets)
    {
        /*
         * first time initialisation. no packets inflight, so we can write
         * at leisure.
         */
        replicate_set_n_buckets(rep, n_buckets);

        if (!REP_HAS_INLINE_BUCKETS(rep))
            vec_validate_aligned(rep->rep_buckets,
                                 rep->rep_n_buckets - 1,
                                 CLIB_CACHE_LINE_BYTES);

        replicate_fill_buckets(rep, nhs,
                               replicate_get_buckets(rep),
                               n_buckets);
    }
    else
    {
        /*
         * This is a modification of an existing replicate.
         * We need to ensure that packets in flight see a consistent state, that
         * is the number of reported buckets the REP has
         * is not more than it actually has. So if the
         * number of buckets is increasing, we must update the bucket array first,
         * then the reported number. vice-versa if the number of buckets goes down.
         */
        if (n_buckets == rep->rep_n_buckets)
        {
            /*
             * no change in the number of buckets. we can simply fill what
             * is new over what is old.
             */
            replicate_fill_buckets(rep, nhs,
                                   replicate_get_buckets(rep),
                                   n_buckets);
        }
        else if (n_buckets > rep->rep_n_buckets)
        {
            /*
             * we have more buckets. the old replicate map (if there is one)
             * will remain valid, i.e. mapping to indices within range, so we
             * update it last.
             */
            if (n_buckets > REP_NUM_INLINE_BUCKETS &&
                rep->rep_n_buckets <= REP_NUM_INLINE_BUCKETS)
            {
                /*
                 * the new increased number of buckets is crossing the threshold
                 * from the inline storage to out-line. Alloc the outline buckets
                 * first, then fixup the number. then reset the inlines.
                 */
                ASSERT(NULL == rep->rep_buckets);
                vec_validate_aligned(rep->rep_buckets,
                                     n_buckets - 1,
                                     CLIB_CACHE_LINE_BYTES);

                replicate_fill_buckets(rep, nhs,
                                       rep->rep_buckets,
                                       n_buckets);
                CLIB_MEMORY_BARRIER();
                replicate_set_n_buckets(rep, n_buckets);

                CLIB_MEMORY_BARRIER();

                for (ii = 0; ii < REP_NUM_INLINE_BUCKETS; ii++)
                {
                    dpo_reset(&rep->rep_buckets_inline[ii]);
                }
            }
            else
            {
                if (n_buckets <= REP_NUM_INLINE_BUCKETS)
                {
                    /*
                     * we are not crossing the threshold and it's still inline buckets.
                     * we can write the new on the old..
                     */
                    replicate_fill_buckets(rep, nhs,
                                           replicate_get_buckets(rep),
                                           n_buckets);
                    CLIB_MEMORY_BARRIER();
                    replicate_set_n_buckets(rep, n_buckets);
                }
                else
                {
                    /*
                     * we are not crossing the threshold. We need a new bucket array to
                     * hold the increased number of choices.
                     */
                    dpo_id_t *new_buckets, *old_buckets, *tmp_dpo;

                    new_buckets = NULL;
                    old_buckets = replicate_get_buckets(rep);

                    vec_validate_aligned(new_buckets,
                                         n_buckets - 1,
                                         CLIB_CACHE_LINE_BYTES);

                    replicate_fill_buckets(rep, nhs, new_buckets, n_buckets);
                    CLIB_MEMORY_BARRIER();
                    rep->rep_buckets = new_buckets;
                    CLIB_MEMORY_BARRIER();
                    replicate_set_n_buckets(rep, n_buckets);

                    vec_foreach(tmp_dpo, old_buckets)
                    {
                        dpo_reset(tmp_dpo);
                    }
                    vec_free(old_buckets);
                }
            }
        }
        else
        {
            /*
             * bucket size shrinkage.
             */
            if (n_buckets <= REP_NUM_INLINE_BUCKETS &&
                rep->rep_n_buckets > REP_NUM_INLINE_BUCKETS)
            {
                /*
                 * the new decreased number of buckets is crossing the threshold
                 * from out-line storage to inline:
                 *   1 - Fill the inline buckets,
                 *   2 - fixup the number (and this point the inline buckets are
                 *       used).
                 *   3 - free the outline buckets
                 */
                replicate_fill_buckets(rep, nhs,
                                       rep->rep_buckets_inline,
                                       n_buckets);
                CLIB_MEMORY_BARRIER();
                replicate_set_n_buckets(rep, n_buckets);
                CLIB_MEMORY_BARRIER();

                vec_foreach(tmp_dpo, rep->rep_buckets)
                {
                    dpo_reset(tmp_dpo);
                }
                vec_free(rep->rep_buckets);
            }
            else
            {
                /*
                 * not crossing the threshold.
                 *  1 - update the number to the smaller size
                 *  2 - write the new buckets
                 *  3 - reset those no longer used.
                 */
                dpo_id_t *buckets;
                u32 old_n_buckets;

                old_n_buckets = rep->rep_n_buckets;
                buckets = replicate_get_buckets(rep);

                replicate_set_n_buckets(rep, n_buckets);
                CLIB_MEMORY_BARRIER();

                replicate_fill_buckets(rep, nhs,
                                       buckets,
                                       n_buckets);

                for (ii = n_buckets; ii < old_n_buckets; ii++)
                {
                    dpo_reset(&buckets[ii]);
                }
            }
        }
    }

    vec_foreach (nh, nhs)
    {
        dpo_reset(&nh->path_dpo);
    }
    vec_free(nhs);
}

static void
replicate_lock (dpo_id_t *dpo)
{
    replicate_t *rep;

    rep = replicate_get(dpo->dpoi_index);

    rep->rep_locks++;
}

index_t
replicate_dup (replicate_flags_t flags,
               index_t repi)
{
    replicate_t *rep, *copy;

    rep = replicate_get(repi);

    if (rep->rep_flags == flags ||
        flags & REPLICATE_FLAGS_HAS_LOCAL)
    {
        /*
         * we can include all the buckets from the original in the copy
         */
        return (repi);
    }
    else
    {
        /*
         * caller doesn't want the local paths that the original has
         */
        if (rep->rep_n_buckets == 1)
        {
            /*
             * original has only one bucket that is the local, so create
             * a new one with only the drop
             */
            copy = replicate_create_i (1, rep->rep_proto);

            replicate_set_bucket_i(copy, 0,
                                   replicate_get_buckets(copy),
                                   drop_dpo_get(rep->rep_proto));
        }
        else
        {
            dpo_id_t *old_buckets, *copy_buckets;
            u16 bucket, pos;

            copy = replicate_create_i(rep->rep_n_buckets - 1,
                                      rep->rep_proto);

            rep = replicate_get(repi);
            old_buckets = replicate_get_buckets(rep);
            copy_buckets = replicate_get_buckets(copy);
            pos = 0;

            for (bucket = 0; bucket < rep->rep_n_buckets; bucket++)
            {
                if (!dpo_is_receive(&old_buckets[bucket]))
                {
                    replicate_set_bucket_i(copy, pos, copy_buckets,
                                           (&old_buckets[bucket]));
                    pos++;
                }
            }
        }
    }

    return (replicate_get_index(copy));
}

static void
replicate_destroy (replicate_t *rep)
{
    dpo_id_t *buckets;
    int i;

    buckets = replicate_get_buckets(rep);

    for (i = 0; i < rep->rep_n_buckets; i++)
    {
        dpo_reset(&buckets[i]);
    }

    REP_DBG(rep, "destroy");
    if (!REP_HAS_INLINE_BUCKETS(rep))
    {
        vec_free(rep->rep_buckets);
    }

    pool_put(replicate_pool, rep);
}

static void
replicate_unlock (dpo_id_t *dpo)
{
    replicate_t *rep;

    rep = replicate_get(dpo->dpoi_index);

    rep->rep_locks--;

    if (0 == rep->rep_locks)
    {
        replicate_destroy(rep);
    }
}

static void
replicate_mem_show (void)
{
    fib_show_memory_usage("replicate",
			  pool_elts(replicate_pool),
			  pool_len(replicate_pool),
			  sizeof(replicate_t));
}

const static dpo_vft_t rep_vft = {
    .dv_lock = replicate_lock,
    .dv_unlock = replicate_unlock,
    .dv_format = format_replicate_dpo,
    .dv_mem_show = replicate_mem_show,
};

/**
 * @brief The per-protocol VLIB graph nodes that are assigned to a replicate
 *        object.
 *
 * this means that these graph nodes are ones from which a replicate is the
 * parent object in the DPO-graph.
 */
const static char* const replicate_ip4_nodes[] =
{
    "ip4-replicate",
    NULL,
};
const static char* const replicate_ip6_nodes[] =
{
    "ip6-replicate",
    NULL,
};
const static char* const replicate_mpls_nodes[] =
{
    "mpls-replicate",
    NULL,
};

const static char* const * const replicate_nodes[DPO_PROTO_NUM] =
{
    [DPO_PROTO_IP4]  = replicate_ip4_nodes,
    [DPO_PROTO_IP6]  = replicate_ip6_nodes,
    [DPO_PROTO_MPLS] = replicate_mpls_nodes,
};

void
replicate_module_init (void)
{
    dpo_register(DPO_REPLICATE, &rep_vft, replicate_nodes);
    replicate_logger = vlib_log_register_class("dpo", "replicate");
}

static clib_error_t *
replicate_show (vlib_main_t * vm,
                unformat_input_t * input,
                vlib_cli_command_t * cmd)
{
    index_t repi = INDEX_INVALID;

    while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
    {
        if (unformat (input, "%d", &repi))
            ;
        else
            break;
    }

    if (INDEX_INVALID != repi)
    {
        vlib_cli_output (vm, "%U", format_replicate, repi,
                         REPLICATE_FORMAT_DETAIL);
    }
    else
    {
        replicate_t *rep;

        pool_foreach(rep, replicate_pool,
        ({
            vlib_cli_output (vm, "%U", format_replicate,
                             replicate_get_index(rep),
                             REPLICATE_FORMAT_NONE);
        }));
    }

    return 0;
}

VLIB_CLI_COMMAND (replicate_show_command, static) = {
    .path = "show replicate",
    .short_help = "show replicate [<index>]",
    .function = replicate_show,
};

typedef struct replicate_trace_t_
{
    index_t rep_index;
    dpo_id_t dpo;
} replicate_trace_t;

static uword
replicate_inline (vlib_main_t * vm,
                  vlib_node_runtime_t * node,
                  vlib_frame_t * frame)
{
    vlib_combined_counter_main_t * cm = &replicate_main.repm_counters;
    replicate_main_t * rm = &replicate_main;
    u32 n_left_from, * from, * to_next, next_index;
    u32 thread_index = vlib_get_thread_index();

    from = vlib_frame_vector_args (frame);
    n_left_from = frame->n_vectors;
    next_index = node->cached_next_index;
  
    while (n_left_from > 0)
    {
        u32 n_left_to_next;

        vlib_get_next_frame (vm, node, next_index,
                             to_next, n_left_to_next);

        while (n_left_from > 0 && n_left_to_next > 0)
	{
            u32 next0, ci0, bi0, bucket, repi0;
            const replicate_t *rep0;
            vlib_buffer_t * b0, *c0;
            const dpo_id_t *dpo0;
	    u8 num_cloned;

            bi0 = from[0];
            from += 1;
            n_left_from -= 1;

            b0 = vlib_get_buffer (vm, bi0);
            repi0 = vnet_buffer (b0)->ip.adj_index[VLIB_TX];
            rep0 = replicate_get(repi0);

            vlib_increment_combined_counter(
                cm, thread_index, repi0, 1,
                vlib_buffer_length_in_chain(vm, b0));

	    vec_validate (rm->clones[thread_index], rep0->rep_n_buckets - 1);

	    num_cloned = vlib_buffer_clone (vm, bi0, rm->clones[thread_index],
                                            rep0->rep_n_buckets,
					    VLIB_BUFFER_CLONE_HEAD_SIZE);

	    if (num_cloned != rep0->rep_n_buckets)
	      {
		vlib_node_increment_counter
		  (vm, node->node_index,
		   REPLICATE_DPO_ERROR_BUFFER_ALLOCATION_FAILURE, 1);
	      }

            for (bucket = 0; bucket < num_cloned; bucket++)
            {
                ci0 = rm->clones[thread_index][bucket];
                c0 = vlib_get_buffer(vm, ci0);

                to_next[0] = ci0;
                to_next += 1;
                n_left_to_next -= 1;

                dpo0 = replicate_get_bucket_i(rep0, bucket);
                next0 = dpo0->dpoi_next_node;
                vnet_buffer (c0)->ip.adj_index[VLIB_TX] = dpo0->dpoi_index;

                if (PREDICT_FALSE(b0->flags & VLIB_BUFFER_IS_TRACED))
                {
                    replicate_trace_t *t;

                    if (c0 != b0)
                    {
                        vlib_buffer_copy_trace_flag (vm, b0, ci0);
                        VLIB_BUFFER_TRACE_TRAJECTORY_INIT (c0);
                    }
                    t = vlib_add_trace (vm, node, c0, sizeof (*t));
                    t->rep_index = repi0;
                    t->dpo = *dpo0;
                }

                vlib_validate_buffer_enqueue_x1 (vm, node, next_index,
                                                 to_next, n_left_to_next,
                                                 ci0, next0);
		if (PREDICT_FALSE (n_left_to_next == 0))
		  {
		    vlib_put_next_frame (vm, node, next_index, n_left_to_next);
		    vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
		  }
            }
	    vec_reset_length (rm->clones[thread_index]);
        }

        vlib_put_next_frame (vm, node, next_index, n_left_to_next);
    }

    return frame->n_vectors;
}

static u8 *
format_replicate_trace (u8 * s, va_list * args)
{
  CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
  CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
  replicate_trace_t *t = va_arg (*args, replicate_trace_t *);

  s = format (s, "replicate: %d via %U",
              t->rep_index,
              format_dpo_id, &t->dpo, 0);
  return s;
}

static uword
ip4_replicate (vlib_main_t * vm,
               vlib_node_runtime_t * node,
               vlib_frame_t * frame)
{
    return (replicate_inline (vm, node, frame));
}

/**
 * @brief IP4 replication node
 */
VLIB_REGISTER_NODE (ip4_replicate_node) = {
  .function = ip4_replicate,
  .name = "ip4-replicate",
  .vector_size = sizeof (u32),

  .n_errors = ARRAY_LEN(replicate_dpo_error_strings),
  .error_strings = replicate_dpo_error_strings,

  .format_trace = format_replicate_trace,
  .n_next_nodes = 1,
  .next_nodes = {
      [0] = "ip4-drop",
  },
};

static uword
ip6_replicate (vlib_main_t * vm,
               vlib_node_runtime_t * node,
               vlib_frame_t * frame)
{
    return (replicate_inline (vm, node, frame));
}

/**
 * @brief IPv6 replication node
 */
VLIB_REGISTER_NODE (ip6_replicate_node) = {
  .function = ip6_replicate,
  .name = "ip6-replicate",
  .vector_size = sizeof (u32),

  .n_errors = ARRAY_LEN(replicate_dpo_error_strings),
  .error_strings = replicate_dpo_error_strings,

  .format_trace = format_replicate_trace,
  .n_next_nodes = 1,
  .next_nodes = {
      [0] = "ip6-drop",
  },
};

static uword
mpls_replicate (vlib_main_t * vm,
                vlib_node_runtime_t * node,
                vlib_frame_t * frame)
{
    return (replicate_inline (vm, node, frame));
}

/**
 * @brief MPLS replication node
 */
VLIB_REGISTER_NODE (mpls_replicate_node) = {
  .function = mpls_replicate,
  .name = "mpls-replicate",
  .vector_size = sizeof (u32),

  .n_errors = ARRAY_LEN(replicate_dpo_error_strings),
  .error_strings = replicate_dpo_error_strings,

  .format_trace = format_replicate_trace,
  .n_next_nodes = 1,
  .next_nodes = {
      [0] = "mpls-drop",
  },
};

clib_error_t *
replicate_dpo_init (vlib_main_t * vm)
{
  replicate_main_t * rm = &replicate_main;

  vec_validate (rm->clones, vlib_num_workers());

  return 0;
}

VLIB_INIT_FUNCTION (replicate_dpo_init);