Aggregator moved to a separate protocol

Also updated data structures and reconfigure.

Known bug: the hash doesn't take adata into account. Needs fixing!

aggr-test.conf

@@ -0,0 +1,116 @@
+log "bird.log" all;
+
+protocol device {}
+
+protocol static {
+  ipv6;
+  route 2001:db8:0::/48 unreachable { bgp_path.prepend(65432); bgp_path.prepend(4200000000); };
+  route 2001:db8:1::/48 unreachable;
+  route 2001:db8:2::/48 unreachable;
+  route 2001:db8:3::/48 unreachable;
+  route 2001:db8:4::/48 unreachable;
+  route 2001:db8:5::/48 unreachable;
+  route 2001:db8:6::/48 unreachable;
+  route 2001:db8:7::/48 unreachable;
+  route 2001:db8:8::/48 unreachable;
+  route 2001:db8:9::/48 unreachable;
+  route 2001:db8:a::/48 unreachable;
+  route 2001:db8:b::/48 unreachable;
+  route 2001:db8:c::/48 unreachable;
+  route 2001:db8:d::/48 unreachable;
+  route 2001:db8:e::/48 unreachable;
+  route 2001:db8:f::/48 unreachable;
+}
+
+protocol static {
+  ipv6 {
+    import filter {
+      bgp_med = 1;
+      bgp_community = -empty-.add((65533,1)).add((65500,0xe));
+      accept;
+    };
+  };
+  route 2001:db8:1::/48 unreachable;
+  route 2001:db8:3::/48 unreachable;
+  route 2001:db8:5::/48 unreachable;
+  route 2001:db8:7::/48 unreachable;
+  route 2001:db8:9::/48 unreachable;
+  route 2001:db8:b::/48 unreachable;
+  route 2001:db8:d::/48 unreachable;
+  route 2001:db8:f::/48 unreachable;
+}
+
+protocol static {
+  ipv6 {
+    import filter {
+      bgp_med = 2;
+      bgp_community = -empty-.add((65533,2)).add((65500,0xd));
+      accept;
+    };
+  };
+  route 2001:db8:2::/48 unreachable;
+  route 2001:db8:3::/48 unreachable;
+  route 2001:db8:6::/48 unreachable;
+  route 2001:db8:7::/48 unreachable;
+  route 2001:db8:a::/48 unreachable;
+  route 2001:db8:b::/48 unreachable;
+  route 2001:db8:e::/48 unreachable;
+  route 2001:db8:f::/48 unreachable;
+}
+
+protocol static {
+  ipv6 {
+    import filter {
+      bgp_med = 4;
+      bgp_community = -empty-.add((65533,4)).add((65500,0xb));
+      accept;
+    };
+  };
+  route 2001:db8:4::/48 unreachable;
+  route 2001:db8:5::/48 unreachable;
+  route 2001:db8:6::/48 unreachable;
+  route 2001:db8:7::/48 unreachable;
+  route 2001:db8:c::/48 unreachable;
+  route 2001:db8:d::/48 unreachable;
+  route 2001:db8:e::/48 unreachable;
+  route 2001:db8:f::/48 unreachable;
+}
+
+protocol static {
+  ipv6 {
+    import filter {
+      bgp_med = 8;
+      bgp_community = -empty-.add((65533,8)).add((65500,0x7));
+      accept;
+    };
+  };
+  route 2001:db8:8::/48 unreachable;
+  route 2001:db8:9::/48 unreachable;
+  route 2001:db8:a::/48 unreachable;
+  route 2001:db8:b::/48 unreachable;
+  route 2001:db8:c::/48 unreachable;
+  route 2001:db8:d::/48 unreachable;
+  route 2001:db8:e::/48 unreachable;
+  route 2001:db8:f::/48 unreachable;
+}
+
+ipv6 table agr_result;
+
+protocol aggregator {
+  table master6;
+  peer table agr_result;
+  export all;
+  aggregate on net,(defined(bgp_med));
+  merge by {
+    print "Merging all these: ", routes;
+    bgp_med = 0;
+    for route r in routes do {
+      if ! defined(r.bgp_med) then { unset(bgp_med); accept; }
+
+      print r, " bgp_med: ", r.bgp_med;
+      bgp_med = bgp_med + r.bgp_med;
+      bgp_community = bgp_community.add(r.bgp_community);
+    }
+    accept;
+  };
+}

conf/confbase.Y

@@ -95,8 +95,7 @@ CF_DECLS
   struct timeformat *tf;
   mpls_label_stack *mls;
   struct bytestring *bs;
-  struct aggr_item *ai;
+  struct aggr_item_node *ai;
-  struct aggr_item_linearized *ail;
 }
 
 %token END CLI_MARKER INVALID_TOKEN ELSECOL DDOT

configure.ac

@@ -320,7 +320,7 @@ if test "$enable_mpls_kernel" != no ; then
   fi
 fi
 
-all_protocols="$proto_bfd babel bgp mrt ospf perf pipe radv rip rpki static"
+all_protocols="aggregator $proto_bfd babel bgp mrt ospf perf pipe radv rip rpki static"
 
 all_protocols=`echo $all_protocols | sed 's/ /,/g'`
 
@@ -328,6 +328,7 @@ if test "$with_protocols" = all ; then
   with_protocols="$all_protocols"
 fi
 
+AH_TEMPLATE([CONFIG_AGGREGATOR],[Aggregator protocol])
 AH_TEMPLATE([CONFIG_BABEL], 	[Babel protocol])
 AH_TEMPLATE([CONFIG_BFD],	[BFD protocol])
 AH_TEMPLATE([CONFIG_BGP],	[BGP protocol])

filter/config.Y

@@ -46,7 +46,7 @@ static inline void f_method_call_start(struct f_inst *object)
     .object = object,
     .main = new_config->current_scope,
     .scope = {
-      .next = NULL,
+      .next = global_root_scope,
       .hash = scope->hash,
       .active = 1,
       .block = 1,
@@ -257,6 +257,8 @@ f_const_empty(enum f_type t)
 	.type = t,
 	.val.ad = &null_adata,
       });
+    case T_ROUTE:
+      return f_new_inst(FI_CONSTANT, (struct f_val) { .type = T_ROUTE });
     default:
       return f_new_inst(FI_CONSTANT, (struct f_val) {});
   }
@@ -876,6 +878,18 @@ method_name_cont:
      $$ = $1->method->new_inst(FM.object, $4);
      f_method_call_end();
    }
+ | static_attr {
+     if (FM.object->type != T_ROUTE)
+       cf_error("Getting a route attribute from %s, need a route", f_type_name(FM.object->type));
+     $$ = f_new_inst(FI_RTA_GET, FM.object, $1);
+     f_method_call_end();
+   }
+ | dynamic_attr {
+     if (FM.object->type != T_ROUTE)
+       cf_error("Getting a route attribute from %s, need a route", f_type_name(FM.object->type));
+     $$ = f_new_inst(FI_EA_GET, FM.object, $1);
+     f_method_call_end();
+   }
  ;
 
 term:
@@ -948,15 +962,6 @@ term:
  | FORMAT '(' term ')' {  $$ = f_new_inst(FI_FORMAT, $3); }
 
  | function_call
-
- | CF_SYM_KNOWN '.' static_attr {
-     cf_assert_symbol($1, SYM_VARIABLE);
-     $$ = f_new_inst(FI_RTA_GET, f_new_inst(FI_VAR_GET, $1), $3);
-   }
- | CF_SYM_KNOWN '.' dynamic_attr {
-     cf_assert_symbol($1, SYM_VARIABLE);
-     $$ = f_new_inst(FI_EA_GET, f_new_inst(FI_VAR_GET, $1), $3);
-   }
  ;
 
 break_command:
@@ -1074,16 +1079,17 @@ lvalue:
      switch ($1->class)
      {
        case SYM_VARIABLE_RANGE:
-	 $$ = (struct f_lval) { .type = F_LVAL_VARIABLE, .sym = $1 };
+	 $$ = (struct f_lval) { .type = F_LVAL_VARIABLE, .sym = $1, .rte = f_const_empty(T_ROUTE) };
          break;
        case SYM_ATTRIBUTE:
-         $$ = (struct f_lval) { .type = F_LVAL_EA, .da = *($1->attribute) };
+         $$ = (struct f_lval) { .type = F_LVAL_EA, .da = *($1->attribute), .rte = f_const_empty(T_ROUTE) };
 	 break;
        default:
 	 cf_error("Variable name or custom attribute name required");
      }
    }
- | static_attr { $$ = (struct f_lval) { .type = F_LVAL_SA, .sa = $1 }; }
+ | static_attr { $$ = (struct f_lval) { .type = F_LVAL_SA, .sa = $1, .rte = f_const_empty(T_ROUTE) }; }
- | dynamic_attr { $$ = (struct f_lval) { .type = F_LVAL_EA, .da = $1 }; };
+ | dynamic_attr { $$ = (struct f_lval) { .type = F_LVAL_EA, .da = $1, .rte = f_const_empty(T_ROUTE) }; }
+ ;
 
 CF_END

filter/data.c

@@ -588,17 +588,17 @@ rte_format(const struct rte *rte, buffer *buf)
 }
 
 static void
-rte_block_format(const struct adata *ad, buffer *buf)
+rte_block_format(const struct rte *rte, buffer *buf)
 {
-  struct rte **routes = ((struct rte_block *)ad)->routes;
+  buffer_print(buf, "Block of routes:");
-  const int count = (ad->length - sizeof(struct rte_block) + sizeof(struct adata)) / sizeof(struct rte *);
 
-  buffer_print(buf, "Block of %d routes:");
+  int i = 0;
-
+  while (rte)
-  for (int i = 0; i < count; i++)
   {
-    buffer_print(buf, "\t%d: ", i);
+    buffer_print(buf, "%s%d: ", i ? "; " : " ", i);
-    rte_format(routes[i], buf);
+    rte_format(rte, buf);
+    rte = rte->next;
+    i++;
   }
 }
 
@@ -631,7 +631,7 @@ val_format(const struct f_val *v, buffer *buf)
   case T_LCLIST: lc_set_format(v->val.ad, -1, buf2, 1000); buffer_print(buf, "(lclist %s)", buf2); return;
   case T_PATH_MASK: pm_format(v->val.path_mask, buf); return;
   case T_ROUTE: rte_format(v->val.rte, buf); return;
-  case T_ROUTES_BLOCK: rte_block_format(v->val.ad, buf); return;
+  case T_ROUTES_BLOCK: rte_block_format(v->val.rte, buf); return;
   default:	buffer_print(buf, "[unknown type %x]", v->type); return;
   }
 }

filter/data.h

@@ -238,11 +238,6 @@ struct f_trie_walk_state
   const struct f_trie_node *stack[TRIE_STACK_LENGTH];
 };
 
-struct rte_block {
-  struct adata ad;
-  struct rte *routes[];
-};
-
 struct f_tree *f_new_tree(void);
 struct f_tree *build_tree(struct f_tree *);
 const struct f_tree *find_tree(const struct f_tree *t, const struct f_val *val);

filter/f-inst.c

@@ -608,8 +608,15 @@
   INST(FI_ROUTES_BLOCK_FOR_NEXT, 3, 0) {
     NEVER_CONSTANT;
     ARG(1, T_ROUTES_BLOCK);
-    if (rte_set_walk(v1.val.ad, &v2.val.i, &v3.val.rte))
+    if (!v2.type)
+      v2 = v1;
+
+    if (v2.val.rte)
+    {
+      v3.val.rte = v2.val.rte;
+      v2.val.rte = v2.val.rte->next;
       LINE(2,0);
+    }
 
     METHOD_CONSTRUCTOR("!for_next");
   }

filter/filter.c

@@ -198,6 +198,7 @@ interpret(struct filter_state *fs, const struct f_line *line, uint argc, const s
 #define v3 vv(2)
 
 #define runtime(fmt, ...) do { \
+  if (what->lineno > 66666) bug("BAD"); \
   if (!(fs->flags & FF_SILENT)) \
     log_rl(&rl_runtime_err, L_ERR "filters, line %d: " fmt, what->lineno, ##__VA_ARGS__); \
   return F_ERROR; \
@@ -349,7 +350,6 @@ f_run_args(const struct filter *filter, struct rte **rte, struct linpool *tmp_po
 enum filter_return
 f_eval_rte(const struct f_line *expr, struct rte **rte, struct linpool *tmp_pool, uint argc, const struct f_val *argv, struct f_val *pres)
 {
-  struct rte *old_rte = *rte;
   filter_state = (struct filter_state) {
     .stack = &filter_stack,
     .rte = rte,
@@ -358,17 +358,7 @@ f_eval_rte(const struct f_line *expr, struct rte **rte, struct linpool *tmp_pool
 
   LOG_BUFFER_INIT(filter_state.buf);
 
-  enum filter_return fret = interpret(&filter_state, expr, argc, argv, pres);
+  return interpret(&filter_state, expr, argc, argv, pres);
-
-  if (filter_state.old_rta || (old_rte != *rte))
-  {
-    ASSERT_DIE(rta_is_cached(filter_state.old_rta) && !rta_is_cached((*rte)->attrs));
-    log(L_WARN "Attempted to change a read-only route, reverting");
-    (*rte)->attrs = filter_state.old_rta;
-    *rte = old_rte;
-  }
-
-  return fret;
 }
 
 /*

nest/Makefile

@@ -1,4 +1,4 @@
-src := a-path.c a-set.c cli.c cmds.c iface.c locks.c neighbor.c password.c proto.c proto-build.c rt-attr.c rt-dev.c rt-fib.c rt-show.c rt-table.c aggregator.c
+src := a-path.c a-set.c cli.c cmds.c iface.c locks.c neighbor.c password.c proto.c proto-build.c rt-attr.c rt-dev.c rt-fib.c rt-show.c rt-table.c
 obj := $(src-o-files)
 $(all-daemon)
 $(cf-local)

nest/a-set.c

@@ -741,19 +741,3 @@ lc_set_walk(const struct adata *list, uint *pos, lcomm *val)
 
   return 1;
 }
-
-int
-rte_set_walk(const struct adata *list, u32 *pos, struct rte **val)
-{
-  if (!list)
-    return 0;
-
-  if (*pos >= (u32)rte_set_get_size(list))
-    return 0;
-
-  struct rte *res = rte_set_get_data(list, *pos);
-  *val = res;
-  *pos += 1;
-
-  return 1;
-}

nest/aggregator.c

@@ -1,672 +0,0 @@
-/*
- *  BIRD Internet Routing Daemon -- Route aggregation
- *
- *  (c) 2023
- *
- *	Can be freely distributed and used under the terms of the GNU GPL.
- */
-
-/**
- * DOC: Route aggregation
- *
- * This is an implementation of route aggregation functionality.
- * It enables user to specify a set of route attributes in the configuarion file
- * and then, for a given destination (net), aggregate routes with the same
- * values of these attributes into a single multi-path route.
- *
- * Structure &channel contains pointer to aggregation list which is represented
- * by &aggr_list_linearized. In rt_notify_aggregated(), attributes from this
- * list are evaluated for every route of a given net and results are stored
- * in &rte_val_list which contains pointer to this route and array of &f_val.
- * Array of pointers to &rte_val_list entries is sorted using
- * sort_rte_val_list(). For comparison of &f_val structures, val_compare()
- * is used. Comparator function is written so that sorting is stable. If all
- * attributes have the same values, routes are compared by their global IDs.
- *
- * After sorting, &rte_val_list entries containing equivalent routes will be
- * adjacent to each other. Function process_rte_list() iterates through these
- * entries to identify sequences of equivalent routes. New route will be
- * created for each such sequence, even if only from a single route.
- * Only attributes from the aggreagation list will be set for the new route.
- * New &rta is created and prepare_rta() is used to copy static and dynamic
- * attributes to new &rta from &rta of the original route. New route is created
- * by create_merged_rte() from new &rta and exported to the routing table.
- */
-
-#undef LOCAL_DEBUG
-
-#ifndef _GNU_SOURCE
-#define _GNU_SOURCE
-#endif
-
-#include "nest/bird.h"
-#include "nest/route.h"
-#include "nest/protocol.h"
-#include "nest/iface.h"
-#include "lib/resource.h"
-#include "lib/event.h"
-#include "lib/timer.h"
-#include "lib/string.h"
-#include "conf/conf.h"
-#include "filter/filter.h"
-#include "filter/data.h"
-#include "lib/hash.h"
-#include "lib/string.h"
-#include "lib/alloca.h"
-#include "lib/flowspec.h"
-#include <stdlib.h>
-
-/* Context of &f_val comparison. */
-struct cmp_ctx {
-  const struct channel *c;
-  const struct network *net;
-  const int val_count;
-  u32 failed:1;
-};
-
-static linpool *rte_update_pool;
-
-/*
- * Set static attribute in @rta from static attribute in @old according to @sa.
- */
-static void
-rta_set_static_attr(struct rta *rta, const struct rta *old, struct f_static_attr sa)
-{
-  switch (sa.sa_code)
-  {
-    case SA_FROM:
-      rta->from = old->from;
-      break;
-
-    case SA_GW:
-      rta->dest = RTD_UNICAST;
-      rta->nh.gw = old->nh.gw;
-      rta->nh.iface = old->nh.iface;
-      rta->nh.next = NULL;
-      rta->hostentry = NULL;
-      rta->nh.labels = 0;
-      break;
-
-    case SA_SCOPE:
-      rta->scope = old->scope;
-      break;
-
-    case SA_DEST:
-      rta->dest = old->dest;
-      rta->nh.gw = IPA_NONE;
-      rta->nh.iface = NULL;
-      rta->nh.next = NULL;
-      rta->hostentry = NULL;
-      rta->nh.labels = 0;
-      break;
-
-    case SA_IFNAME:
-      rta->dest = RTD_UNICAST;
-      rta->nh.gw = IPA_NONE;
-      rta->nh.iface = old->nh.iface;
-      rta->nh.next = NULL;
-      rta->hostentry = NULL;
-      rta->nh.labels = 0;
-      break;
-
-    case SA_GW_MPLS:
-      rta->nh.labels = old->nh.labels;
-      memcpy(&rta->nh.label, &old->nh.label, sizeof(u32) * old->nh.labels);
-      break;
-
-    case SA_WEIGHT:
-      rta->nh.weight = old->nh.weight;
-      break;
-
-    case SA_PREF:
-      rta->pref = old->pref;
-      break;
-
-    default:
-      bug("Invalid static attribute access (%u/%u)", sa.f_type, sa.sa_code);
-  }
-}
-
-static int
-get_dynamic_attr_count(const struct aggr_item_linearized *ail)
-{
-  int ea_count = 0;
-
-  for (int i = 0; i < ail->count; i++)
-    if (ail->items[i].type == AGGR_ITEM_DYNAMIC_ATTR)
-      ea_count++;
-
-  return ea_count;
-}
-
-/*
- * Copy static and dynamic attributes from @old to @new according to
- * aggregation list @ail. Because route may not have any extended
- * attributes, return real number of attributes that were copied.
- */
-static int
-prepare_rta(struct rta *new, const struct rta *old, const struct aggr_item_linearized *ail)
-{
-  int pos = 0;
-
-  for (int i = 0; i < ail->count; i++)
-  {
-    if (ail->items[i].type == AGGR_ITEM_DYNAMIC_ATTR)
-    {
-      u32 ea_code = ail->items[i].da.ea_code;
-      const struct eattr *e = ea_find(old->eattrs, ea_code);
-
-      if (e)
-        new->eattrs->attrs[pos++] = *e;
-    }
-    else if (ail->items[i].type == AGGR_ITEM_STATIC_ATTR)
-      rta_set_static_attr(new, old, ail->items[i].sa);
-  }
-
-  return pos;
-}
-
-/*
- * Find route with lowest ID in a sequence of rte_val_list entries
- * within range [start, start + length).
- * @start: first element in a sequence of equivalent routes
- */
-static const struct rte *
-find_rte_lowest_id(const struct rte_val_list **start, int length)
-{
-  const struct rte *rte = start[0]->rte;
-  u32 id = rte->src->global_id;
-
-  for (int i = 1; i < length; i++)
-  {
-    u32 current = start[i]->rte->src->global_id;
-
-    if (current < id)
-    {
-      id = current;
-      rte = start[i]->rte;
-    }
-  }
-
-  log("Lowest ID: %d", id);
-  return rte;
-}
-
-static int
-compare_f_val(const struct f_val *v1, const struct f_val *v2, struct cmp_ctx *ctx)
-{
-  int result = val_compare(v1, v2);
-
-  if (result != F_CMP_ERROR)
-    return result;
-
-  ctx->failed = 1;
-
-  struct buffer buf;
-  LOG_BUFFER_INIT(buf);
-
-  buffer_puts(&buf, "v1 = ");
-  val_format(v1, &buf);
-  buffer_puts(&buf, ", v2 = ");
-  val_format(v2, &buf);
-  log(L_WARN "%s.%s: Error comparing values while aggregating routes to %N: %s",
-      ctx->c->proto->name, ctx->c->name, ctx->net->n.addr, buf.start);
-
-  bug("Sorting routes according to aggregation list: F_CMP_ERROR");
-}
-
-/*
- * Compare list of &f_val entries.
- * @count: number of &f_val entries
- */
-static int
-compare_val_list(const struct f_val *v1, const struct f_val *v2, struct cmp_ctx *ctx)
-{
-  for (int i = 0; i < ctx->val_count; i++)
-  {
-    int res = compare_f_val(&v1[i], &v2[i], ctx);
-    if (res != 0)
-      return res;
-  }
-
-  return 0;
-}
-
-/*
- * Comparator function for sorting array of pointers to &rte_val_list structures.
- * Compare lists of &f_val associated with routes.
- * If all values are equal, compare route's global IDs.
- * @count: pointer to number of f_val entries
- */
-static int
-compare_val_list_id(const void *fst, const void *snd, void *context)
-{
-  struct cmp_ctx *ctx = (struct cmp_ctx *)context;
-
-  for (int i = 0; i < ctx->val_count; i++)
-  {
-    /*
-     * This function receives two void pointers.
-     * Since we are sorting array of pointers, we have to cast this void
-     * pointer to struct rte_val_list** (pointer to first array element,
-     * which is a pointer). Dereference it once to access this element,
-     * which is struct rte_val_list*. Finally access f_val at position i
-     * and take its address, thus getting struct f_val*.
-     */
-    const struct f_val *v1 = &(*(struct rte_val_list **)fst)->values[i];
-    const struct f_val *v2 = &(*(struct rte_val_list **)snd)->values[i];
-    int result = compare_f_val(v1, v2, ctx);
-
-    if (result != 0)
-      return result;
-  }
-
-  u32 id1 = (*(struct rte_val_list **)fst)->rte->src->global_id;
-  u32 id2 = (*(struct rte_val_list **)snd)->rte->src->global_id;
-  return id1 < id2 ? -1 : 1;
-}
-
-/*
- * Sort array of pointers to &rte_val_list entries.
- * @rte_val: first element in array of pointers to &rte_val_list
- * @rte_count: number of &rte_val_list entries
- * @val_count: number of &f_val entries in each &rte_val_list entry
- */
-static void
-sort_rte_val_list(const struct rte_val_list **rte_val, int rte_count, struct cmp_ctx *ctx)
-{
-  log("======== Sorting routes... ========");
-  qsort_r(rte_val, rte_count, sizeof(struct rte_val_list *), compare_val_list_id, (void *)ctx);
-
-  for (int i = 0; i < rte_count; i++)
-    log("route ID: %d", rte_val[i]->rte->src->global_id);
-}
-
-/*
- * Create and export new merged route.
- * @old: first route in a sequence of equivalent routes that are to be merged
- * @rte_val: first element in a sequence of equivalent rte_val_list entries
- * @length: number of equivalent routes that are to be merged (at least 1)
- * @ail: aggregation list
- */
-static void
-create_merged_rte(struct channel *c, struct network *net, const struct rte_val_list **rte_val,
-                  int length, const struct aggr_item_linearized *ail, int refeed)
-{
-  const struct rte *old = rte_val[0]->rte;
-  const struct rta *rta_old = old->attrs;
-  struct rta *rta_new = allocz(rta_size(rta_old));
-
-  int ea_count = get_dynamic_attr_count(ail);
-  struct ea_list *eal = allocz(sizeof(struct ea_list) + sizeof(struct eattr) * ea_count);
-
-  rta_new->dest = RTD_UNREACHABLE;
-  rta_new->eattrs = eal;
-  eal->next = NULL;
-  eal->count = prepare_rta(rta_new, rta_old, ail);
-
-  const struct rte *rte_lowest = find_rte_lowest_id(rte_val, length);
-  struct rte *new = rte_get_temp(rta_new, rte_lowest->src);
-  new->net = net;
-
-  do_rt_notify(c, net, new, NULL, refeed);
-  log("=============== CREATE MERGED ROUTE ===============");
-  log("New route created: id = %d, protocol: %s", new->src->global_id, new->src->proto->name);
-  log("===================================================");
-
-  struct rte_block *rb = allocz(sizeof(struct rte_block) + sizeof(struct rte *) * length);
-  rb->ad.length = sizeof(struct rte_block) + sizeof(struct rte *) * length - sizeof(struct adata);
-
-  for (int i = 0; i < length; i++)
-    rb->routes[i] = (struct rte *)rte_val[i]->rte;
-
-  /* merge filter needs one argument called "routes" */
-  struct f_val val = {
-    .type = T_ROUTES_BLOCK,
-    .val.ad = &rb->ad,
-  };
-
-  f_eval_rte(ail->merge_filter, &new, rte_update_pool, 1, &val, 0);
-}
-
-/*
- * Iterate through &rte_val_list entries and identify all sequences of
- * equivalent routes.
- * @rte_count: total number of routes being processed
- * @val_count: number of &f_val entries with each route
- * @ail: aggregation list
- */
-static void
-process_rte_list(struct channel *c, struct network *net, const struct rte_val_list **rte_val,
-                 int rte_count, int val_count, const struct aggr_item_linearized *ail, int refeed)
-{
-  if (rte_count == 1)
-  {
-    create_merged_rte(c, net, rte_val, 1, ail, refeed);
-    return;
-  }
-
-  struct cmp_ctx ctx = {
-    .c = c,
-    .net = net,
-    .val_count = val_count,
-    .failed = 0,
-  };
-
-  /*
-   * &start and &current are initially indices to first and second of
-   * &rte_val_list entries. If these entries contain equivalent routes,
-   * &current is incremented until non-equivalent route is found.
-   * [start, current) then define a range of routes that are to be merged.
-   * When non-equivalent route is found, &start is updated and the process
-   * continues until all entries are processed.
-   */
-  int start = 0;
-  int current = 1;
-  log("RTE count: %d", rte_count);
-  log("VAL count: %d", val_count);
-
-  while (start < rte_count && current < rte_count)
-  {
-    int res = compare_val_list(&rte_val[start]->values[0], &rte_val[current]->values[0], &ctx);
-
-    /* At least two equivalent routes were found, try to find more. */
-    if (res == 0)
-    {
-      int merged = 1;
-
-      while (current < rte_count && res == 0)
-      {
-        log("Routes %d and %d are equal", rte_val[start]->rte->src->global_id, rte_val[current]->rte->src->global_id);
-        current++;
-        merged++;
-
-        if (current < rte_count)
-          res = compare_val_list(&rte_val[start]->values[0], &rte_val[current]->values[0], &ctx);
-      }
-
-      log("Creating merged route from %d routes", merged);
-      create_merged_rte(c, net, &rte_val[start], merged, ail, refeed);
-      start = current;
-      current++;
-    }
-    else
-    {
-      log("Route %d and %d are NOT equal", rte_val[start]->rte->src->global_id, rte_val[current]->rte->src->global_id);
-      log("Creating standalone route from route %d", rte_val[start]->rte->src->global_id);
-      create_merged_rte(c, net, &rte_val[start], 1, ail, refeed);
-      start = current;
-      current++;
-    }
-  }
-
-  if (start < rte_count)
-  {
-    log("Creating standalone route from route %d", rte_val[start]->rte->src->global_id);
-    create_merged_rte(c, net, &rte_val[start], 1, ail, refeed);
-  }
-}
-
-static int
-get_rte_count(const struct rte *rte)
-{
-  int count = 0;
-  for (; rte; rte = rte->next)
-    count++;
-  return count;
-}
-
-static void
-log_attributes(const struct f_val *val, int count)
-{
-    struct buffer buf;
-    LOG_BUFFER_INIT(buf);
-
-    for (int i = 0; i < count; i++)
-    {
-      val_format(&val[i], &buf);
-      log("%s", buf.start);
-    }
-}
-
-/*
- * Evaluate static attribute of @rt1 according to @sa
- * and store result in @pos.
- */
-static void
-eval_static_attr(const struct rte *rt1, struct f_static_attr sa, struct f_val *pos)
-{
-  const struct rta *rta = rt1->attrs;
-
-#define RESULT(_type, value, result)    \
-  do {                                  \
-    pos->type = _type;                  \
-    pos->val.value = result;            \
-  } while (0)
-
-  switch (sa.sa_code)
-  {
-    case SA_FROM:       RESULT(sa.f_type, ip, rta->from); break;
-    case SA_GW:	        RESULT(sa.f_type, ip, rta->nh.gw); break;
-    case SA_PROTO:	    RESULT(sa.f_type, s, rt1->src->proto->name); break;
-    case SA_SOURCE:	    RESULT(sa.f_type, i, rta->source); break;
-    case SA_SCOPE:	    RESULT(sa.f_type, i, rta->scope); break;
-    case SA_DEST:	    RESULT(sa.f_type, i, rta->dest); break;
-    case SA_IFNAME:	    RESULT(sa.f_type, s, rta->nh.iface ? rta->nh.iface->name : ""); break;
-    case SA_IFINDEX:	RESULT(sa.f_type, i, rta->nh.iface ? rta->nh.iface->index : 0); break;
-    case SA_WEIGHT:	    RESULT(sa.f_type, i, rta->nh.weight + 1); break;
-    case SA_PREF:	    RESULT(sa.f_type, i, rta->pref); break;
-    case SA_GW_MPLS:    RESULT(sa.f_type, i, rta->nh.labels ? rta->nh.label[0] : MPLS_NULL); break;
-    default:
-      bug("Invalid static attribute access (%u/%u)", sa.f_type, sa.sa_code);
-  }
-
-#undef RESULT
-}
-
-/*
- * Evaluate dynamic attribute of @rt1 according to @da
- * and store result in @pos.
- */
-static void
-eval_dynamic_attr(const struct rte *rt1, struct f_dynamic_attr da, struct f_val *pos)
-{
-  const struct rta *rta = rt1->attrs;
-  const struct eattr *e = ea_find(rta->eattrs, da.ea_code);
-
-#define RESULT(_type, value, result)    \
-  do {                                  \
-    pos->type = _type;                  \
-    pos->val.value = result;            \
-  } while (0)
-
-#define RESULT_VOID         \
-  do {                      \
-    pos->type = T_VOID;     \
-  } while (0)
-
-  if (!e)
-  {
-    /* A special case: undefined as_path looks like empty as_path */
-    if (da.type == EAF_TYPE_AS_PATH)
-    {
-      RESULT(T_PATH, ad, &null_adata);
-      return;
-    }
-
-    /* The same special case for int_set */
-    if (da.type == EAF_TYPE_INT_SET)
-    {
-      RESULT(T_CLIST, ad, &null_adata);
-      return;
-    }
-
-    /* The same special case for ec_set */
-    if (da.type == EAF_TYPE_EC_SET)
-    {
-      RESULT(T_ECLIST, ad, &null_adata);
-      return;
-    }
-
-    /* The same special case for lc_set */
-    if (da.type == EAF_TYPE_LC_SET)
-    {
-      RESULT(T_LCLIST, ad, &null_adata);
-      return;
-    }
-
-    /* Undefined value */
-    RESULT_VOID;
-    return;
-  }
-
-  switch (e->type & EAF_TYPE_MASK)
-  {
-    case EAF_TYPE_INT:
-      RESULT(da.f_type, i, e->u.data);
-      break;
-    case EAF_TYPE_ROUTER_ID:
-      RESULT(T_QUAD, i, e->u.data);
-      break;
-    case EAF_TYPE_OPAQUE:
-      RESULT(T_ENUM_EMPTY, i, 0);
-      break;
-    case EAF_TYPE_IP_ADDRESS:
-      RESULT(T_IP, ip, *((ip_addr *) e->u.ptr->data));
-      break;
-    case EAF_TYPE_AS_PATH:
-      RESULT(T_PATH, ad, e->u.ptr);
-      break;
-    case EAF_TYPE_BITFIELD:
-      RESULT(T_BOOL, i, !!(e->u.data & (1u << da.bit)));
-      break;
-    case EAF_TYPE_INT_SET:
-      RESULT(T_CLIST, ad, e->u.ptr);
-      break;
-    case EAF_TYPE_EC_SET:
-      RESULT(T_ECLIST, ad, e->u.ptr);
-      break;
-    case EAF_TYPE_LC_SET:
-      RESULT(T_LCLIST, ad, e->u.ptr);
-      break;
-    default:
-      bug("Unknown dynamic attribute type");
-  }
-
-#undef RESULT
-#undef RESULT_VOID
-}
-
-void
-rt_notify_aggregated(struct channel *c, struct network *net, struct rte *new_changed, struct rte *old_changed,
-		             struct rte *new_best, struct rte *old_best, int refeed)
-{
-  const struct aggr_item_linearized *ail = c->ai_aggr;
-  const int attr_count = ail->count;
-
-  if (net->routes == NULL)
-    return;
-
-  struct rte *best0 = net->routes;
-  const int rte_count = get_rte_count(best0);
-
-  if (rte_count == 0)
-    return;
-
-  log("---- RT NOTIFY AGGREGATED ----");
-  log("Routes count: %d", rte_count);
-  log("Aggregation list attributes count: %d", attr_count);
-  log("aggr_item_linearized: %p", ail);
-
-  struct rte **rte_temp = allocz(sizeof(struct rte *) * rte_count);
-  struct rte **rte_free_temp = allocz(sizeof(struct rte *) * rte_count);
-
-  int rte_temp_count = 0;
-  int rte_free_count = 0;
-
-  /* Run filter for all routes before aggregation. */
-  for (struct rte *rt0 = best0; rt0; rt0 = rt0->next)
-  {
-    struct rte *rte_free = NULL;
-    struct rte *filtered = export_filter(c, rt0, &rte_free, 0);
-
-    if (filtered)
-      rte_temp[rte_temp_count++] = filtered;
-
-    if (rte_free)
-      rte_free_temp[rte_free_count++] = rte_free;
-  }
-
-  const struct rte_val_list **rte_val_list_ptrs = allocz(sizeof(struct rte_val_list *) * rte_count);
-  int rte_val_list_pos = 0;
-
-  for (int rte_idx = 0; rte_idx < rte_temp_count; rte_idx++)
-  {
-    struct rte *rt0 = rte_temp[rte_idx];
-    struct rte_val_list *rte_val = allocz(sizeof(struct rte_val_list) + sizeof(struct f_val) * attr_count);
-
-    rte_val->rte = rt0;
-    rte_val_list_ptrs[rte_val_list_pos++] = rte_val;
-
-    for (int val_idx = 0; val_idx < attr_count; val_idx++)
-    {
-      int type = ail->items[val_idx].type;
-
-      /* Evaluate route attributes. */
-      switch (type)
-      {
-        case AGGR_ITEM_TERM: {
-          const struct f_line *line = ail->items[val_idx].line;
-          struct rte *rt1 = rt0;
-          enum filter_return fret = f_eval_rte(line, &rt1, rte_update_pool, 0, NULL, &rte_val->values[val_idx]);
-
-          if (rt1 != rt0)
-          {
-            rte_free(rt1);
-            log(L_WARN "rt1 != rt0");
-          }
-
-          if (fret > F_RETURN)
-            log(L_WARN "%s.%s: Wrong number of items left on stack after evaluation of aggregation list", rt1->src->proto->name, rt1->sender);
-
-          break;
-        }
-
-        case AGGR_ITEM_STATIC_ATTR: {
-          struct f_val *pos = &rte_val->values[val_idx];
-          eval_static_attr(rt0, ail->items[val_idx].sa, pos);
-          break;
-        }
-
-        case AGGR_ITEM_DYNAMIC_ATTR: {
-          struct f_val *pos = &rte_val->values[val_idx];
-          eval_dynamic_attr(rt0, ail->items[val_idx].da, pos);
-          break;
-        }
-
-        default:
-          break;
-      }
-    }
-
-    log_attributes(&rte_val->values[0], attr_count);
-  }
-
-  struct cmp_ctx ctx = {
-    .c = c,
-    .net = net,
-    .val_count = attr_count,
-    .failed = 0,
-  };
-
-  sort_rte_val_list(rte_val_list_ptrs, rte_temp_count, &ctx);
-
-  if (ctx.failed)
-    log(L_WARN "%s.%s: Could not aggregate routes to %N due to previous errors", c->proto->name, c->name, net->n.addr);
-  else
-    process_rte_list(c, net, rte_val_list_ptrs, rte_temp_count, attr_count, ail, refeed);
-
-  for (int i = 0; i < rte_free_count; i++)
-    rte_free(rte_free_temp[i]);
-}
-

nest/attrs.h

@@ -140,15 +140,9 @@ static inline int ec_set_get_size(const struct adata *list)
 static inline int lc_set_get_size(const struct adata *list)
 { return list->length / 12; }
 
-static inline int rte_set_get_size(const struct adata *list)
-{ return list->length / sizeof(struct rte *); }
-
 static inline u32 *int_set_get_data(const struct adata *list)
 { return (u32 *) list->data; }
 
-static inline struct rte *rte_set_get_data(const struct adata *list, u32 idx)
-{ return ((struct rte_block *)list)->routes[idx]; }
-
 static inline u32 ec_hi(u64 ec) { return ec >> 32; }
 static inline u32 ec_lo(u64 ec) { return ec; }
 

nest/config.Y

@@ -16,7 +16,6 @@ CF_HDR
 #include "lib/mac.h"
 
 CF_DEFINES
-#define AGGR_ITEM_ALLOC cfg_allocz(sizeof(struct aggr_item))
 
 static struct rtable_config *this_table;
 static struct proto_config *this_proto;
@@ -127,7 +126,6 @@ CF_KEYWORDS(GRACEFUL, RESTART, WAIT, MAX, AS)
 CF_KEYWORDS(MIN, IDLE, RX, TX, INTERVAL, MULTIPLIER, PASSIVE)
 CF_KEYWORDS(CHECK, LINK)
 CF_KEYWORDS(SORTED, TRIE, MIN, MAX, SETTLE, TIME, GC, THRESHOLD, PERIOD)
-CF_KEYWORDS(AGGREGATE, ON, MERGE, BY)
 
 /* For r_args_channel */
 CF_KEYWORDS(IPV4, IPV4_MC, IPV4_MPLS, IPV6, IPV6_MC, IPV6_MPLS, IPV6_SADR, VPN4, VPN4_MC, VPN4_MPLS, VPN6, VPN6_MC, VPN6_MPLS, ROA4, ROA6, FLOW4, FLOW6, MPLS, PRI, SEC)
@@ -141,7 +139,6 @@ CF_ENUM_PX(T_ENUM_AF, AF_, AFI_, IPV4, IPV6)
 
 %type <i32> idval
 %type <f> imexport
-%type <fl> merge_filter
 %type <r> rtable
 %type <s> optproto
 %type <ra> r_args
@@ -154,8 +151,6 @@ CF_ENUM_PX(T_ENUM_AF, AF_, AFI_, IPV4, IPV6)
 %type <net_ptr> r_args_for
 %type <t> channel_sym
 %type <c> channel_arg
-%type <ai> aggr_item aggr_list
-%type <ail> aggr_definition
 
 CF_GRAMMAR
 
@@ -303,121 +298,6 @@ proto_item:
  ;
 
 
-aggr_list:
-   aggr_item
- | aggr_list ',' aggr_item {
-       if ($3 == NULL) {
-         $$ = $1;
-       } else {
-         $$ = $3;
-         $$->next = $1;
-       }
-   }
- ;
-
-aggr_item:
-   '(' term ')' {
-       $$ = AGGR_ITEM_ALLOC;
-       $$->internal.type = AGGR_ITEM_TERM;
-       $$->internal.line = f_linearize($2, 1);
-    }
-  | CF_SYM_KNOWN {
-      switch ($1->class) {
-        case SYM_ATTRIBUTE:
-          $$ = AGGR_ITEM_ALLOC;
-          $$->internal.type = AGGR_ITEM_DYNAMIC_ATTR;
-          $$->internal.da = *$1->attribute;
-          break;
-        case SYM_CONSTANT_RANGE:
-          $$ = NULL;
-          break;
-        default:
-          cf_error("Can't aggregate on symbol type %s.", cf_symbol_class_name($1));
-      }
-    }
-  | dynamic_attr {
-      $$ = AGGR_ITEM_ALLOC;
-      $$->internal.type = AGGR_ITEM_DYNAMIC_ATTR;
-      $$->internal.da = $1;
-    }
-  | static_attr {
-      $$ = AGGR_ITEM_ALLOC;
-      $$->internal.type = AGGR_ITEM_STATIC_ATTR;
-      $$->internal.sa = $1;
-    }
-  ;
-
-merge_filter: MERGE BY {
-  cf_push_block_scope(new_config);
-  cf_create_symbol(new_config, "routes", SYM_VARIABLE | T_ROUTES_BLOCK, offset, f_new_var(sym_->scope));
-} function_body {
-  cf_pop_block_scope(new_config);
-  $4->args++;
-  $$ = $4;
-  }
- ;
-
-aggr_definition:
-    AGGREGATE ON aggr_list merge_filter {
-      _Bool net_present = 0;
-      int count = 0;
-
-      for (const struct aggr_item *item = $3; item; item = item->next) {
-        if (item->internal.type == AGGR_ITEM_STATIC_ATTR && item->internal.sa.sa_code == SA_NET) {
-          net_present = 1;
-          continue;
-        }
-
-        count++;
-      }
-
-     if (!net_present)
-       cf_error("'NET' must be present");
-
-     size_t allocated = sizeof(struct aggr_item_linearized) + sizeof(struct aggr_item_internal) * count;
-     struct aggr_item_linearized *linear = cfg_allocz(allocated);
-
-     log("nest/config.Y: linear: %p, allocated: %d, count: %d", linear, allocated, count);
-
-     int pos = 0;
-     for (const struct aggr_item *item = $3; item; item = item->next) {
-       if (item->internal.type == AGGR_ITEM_STATIC_ATTR && item->internal.sa.sa_code == SA_NET)
-         continue;
-
-       linear->items[pos++] = item->internal;
-     }
-
-     linear->count = pos;
-     linear->merge_filter = $4;
-
-     int node = 1;
-
-     if (!linear) {
-       for (int i = 0; i < linear->count; i++) {
-         switch (linear->items[i].type) {
-           case AGGR_ITEM_TERM:
-             log("node %d, type: term", node);
-             break;
-           case AGGR_ITEM_STATIC_ATTR:
-             log("node %d, type: static", node);
-             break;
-           case AGGR_ITEM_DYNAMIC_ATTR:
-             log("node %d, type: dynamic", node);
-             break;
-           default:
-             log("node %d, type: other", node);
-             break;
-         }
-
-         node++;
-       }
-     }
-
-     $$ = linear;
-  }
-  ;
-
-
 channel_start: net_type
 {
   $$ = this_channel = channel_config_get(NULL, net_label[$1], $1, this_proto);
@@ -431,7 +311,6 @@ channel_item_:
    }
  | IMPORT imexport { this_channel->in_filter = $2; }
  | EXPORT imexport { this_channel->out_filter = $2; }
- | EXPORT imexport aggr_definition { this_channel->out_filter = $2; this_channel->ai_aggr = $3; this_channel->ra_mode = RA_AGGREGATED; }
  | RECEIVE LIMIT limit_spec { this_channel->rx_limit = $3; }
  | IMPORT LIMIT limit_spec { this_channel->in_limit = $3; }
  | EXPORT LIMIT limit_spec { this_channel->out_limit = $3; }

nest/proto.c

@@ -90,41 +90,6 @@ proto_log_state_change(struct proto *p)
     p->last_state_name_announced = NULL;
 }
 
-static int
-aggr_item_same(const struct aggr_item_internal *fst, const struct aggr_item_internal *snd)
-{
-  if (fst->type != snd->type)
-    return 0;
-
-  switch (fst->type)
-  {
-    case AGGR_ITEM_TERM:
-      return f_same(fst->line, snd->line);
-    case AGGR_ITEM_STATIC_ATTR:
-      return memcmp(&fst->sa, &snd->sa, sizeof(struct f_static_attr)) == 0;
-    case AGGR_ITEM_DYNAMIC_ATTR:
-      return memcmp(&fst->da, &snd->da, sizeof(struct f_dynamic_attr)) == 0;
-    default:
-      bug("Broken aggregating data");
-  }
-}
-
-static int
-aggr_item_linearized_same(const struct aggr_item_linearized *fst, const struct aggr_item_linearized *snd)
-{
-  if (!fst || !snd)
-    return 0;
-
-  if (fst->count != snd->count)
-    return 0;
-
-  for (int i = 0; i < fst->count; i++)
-    if (!aggr_item_same(&fst->items[i], &snd->items[i]))
-      return 0;
-
-  return 1;
-}
-
 struct channel_config *
 proto_cf_find_channel(struct proto_config *pc, uint net_type)
 {
@@ -202,7 +167,6 @@ proto_add_channel(struct proto *p, struct channel_config *cf)
 
   c->in_filter = cf->in_filter;
   c->out_filter = cf->out_filter;
-  c->ai_aggr = cf->ai_aggr;
   c->rx_limit = cf->rx_limit;
   c->in_limit = cf->in_limit;
   c->out_limit = cf->out_limit;
@@ -876,13 +840,6 @@ channel_reconfigure(struct channel *c, struct channel_config *cf)
   int import_changed = !filter_same(cf->in_filter, c->in_filter);
   int export_changed = !filter_same(cf->out_filter, c->out_filter);
 
-  if (cf->ra_mode == RA_AGGREGATED)
-  {
-    export_changed |= !aggr_item_linearized_same(cf->ai_aggr, c->ai_aggr);
-    export_changed |= !f_same(cf->ai_aggr->merge_filter, c->ai_aggr->merge_filter);
-    c->ai_aggr = cf->ai_aggr;
-  }
-
   int rpki_reload_changed = (cf->rpki_reload != c->rpki_reload);
 
   if (c->preference != cf->preference)
@@ -895,7 +852,6 @@ channel_reconfigure(struct channel *c, struct channel_config *cf)
   c->in_filter = cf->in_filter;
   c->out_filter = cf->out_filter;
   c->rx_limit = cf->rx_limit;
-  c->ai_aggr = cf->ai_aggr;
   c->in_limit = cf->in_limit;
   c->out_limit = cf->out_limit;
 

nest/protocol.h

@@ -39,6 +39,7 @@ struct symbol;
 
 enum protocol_class {
   PROTOCOL_NONE,
+  PROTOCOL_AGGREGATOR,
   PROTOCOL_BABEL,
   PROTOCOL_BFD,
   PROTOCOL_BGP,
@@ -103,7 +104,7 @@ void protos_dump_all(void);
 
 extern struct protocol
   proto_device, proto_radv, proto_rip, proto_static, proto_mrt,
-  proto_ospf, proto_perf,
+  proto_ospf, proto_perf, proto_aggregator,
   proto_pipe, proto_bgp, proto_bmp, proto_bfd, proto_babel, proto_rpki;
 
 /*
@@ -487,7 +488,6 @@ struct channel_config {
   struct proto_config *parent;		/* Where channel is defined (proto or template) */
   struct rtable_config *table;		/* Table we're attached to */
   const struct filter *in_filter, *out_filter; /* Attached filters */
-  const struct aggr_item_linearized *ai_aggr;
   struct channel_limit rx_limit;	/* Limit for receiving routes from protocol
 					   (relevant when in_keep_filtered is active) */
   struct channel_limit in_limit;	/* Limit for importing routes from protocol */
@@ -514,7 +514,6 @@ struct channel {
   struct rtable *table;
   const struct filter *in_filter;	/* Input filter */
   const struct filter *out_filter;	/* Output filter */
-  const struct aggr_item_linearized *ai_aggr;
   struct bmap export_map;		/* Keeps track which routes passed export filter */
   struct channel_limit rx_limit;	/* Receive limit (for in_keep_filtered) */
   struct channel_limit in_limit;	/* Input limit */

nest/route.h

@@ -286,7 +286,6 @@ static inline int rte_is_filtered(rte *r) { return !!(r->flags & REF_FILTERED);
 #define RA_ACCEPTED	2		/* Announcement of first accepted route */
 #define RA_ANY		3		/* Announcement of any route change */
 #define RA_MERGED	4		/* Announcement of optimal route merged with next ones */
-#define RA_AGGREGATED 5     /* Announcement of merged routes */
 
 /* Return value of preexport() callback */
 #define RIC_ACCEPT	1		/* Accepted by protocol */
@@ -325,7 +324,6 @@ void rte_update2(struct channel *c, const net_addr *n, rte *new, struct rte_src
 /* rte_update() moved to protocol.h to avoid dependency conflicts */
 int rt_examine(rtable *t, net_addr *a, struct channel *c, const struct filter *filter);
 rte *rt_export_merged(struct channel *c, net *net, rte **rt_free, linpool *pool, int silent);
-rte *rt_export_aggregated(struct channel *c, net *net, rte **rt_free, linpool *pool, int silent);
 void rt_refresh_begin(rtable *t, struct channel *c);
 void rt_refresh_end(rtable *t, struct channel *c);
 void rt_modify_stale(rtable *t, struct channel *c);
@@ -346,6 +344,8 @@ void rt_prune_sync(rtable *t, int all);
 int rte_update_out(struct channel *c, const net_addr *n, rte *new, rte *old0, int refeed);
 struct rtable_config *rt_new_table(struct symbol *s, uint addr_type);
 
+int rte_same(rte *x, rte *y);
+
 static inline int rt_is_ip(rtable *tab)
 { return (tab->addr_type == NET_IP4) || (tab->addr_type == NET_IP6); }
 
@@ -478,7 +478,8 @@ typedef struct rta {
 #define RTS_BABEL 13			/* Babel route */
 #define RTS_RPKI 14			/* Route Origin Authorization */
 #define RTS_PERF 15			/* Perf checker */
-#define RTS_MAX 16
+#define RTS_AGGREGATED 16		/* Aggregated route */
+#define RTS_MAX 17
 
 #define RTD_NONE 0			/* Undefined next hop */
 #define RTD_UNICAST 1			/* Next hop is neighbor router */
@@ -759,44 +760,11 @@ int rt_flowspec_check(rtable *tab_ip, rtable *tab_flow, const net_addr *n, rta *
 #define ROA_VALID	1
 #define ROA_INVALID	2
 
-/*
- * Aggregating routes
- */
-
-enum aggr_item_type {
-  AGGR_ITEM_TERM,
-  AGGR_ITEM_STATIC_ATTR,
-  AGGR_ITEM_DYNAMIC_ATTR,
-};
-
-struct aggr_item_internal {
-  enum aggr_item_type type;
-  union {
-    struct f_static_attr sa;
-    struct f_dynamic_attr da;
-    const struct f_line *line;
-  };
-};
-
-struct aggr_item {
-  const struct aggr_item *next;
-  struct aggr_item_internal internal;
-};
-
-struct aggr_item_linearized {
-  int count;
-  const struct f_line *merge_filter;
-  struct aggr_item_internal items[];
-};
-
 struct rte_val_list {
   const struct rte *rte;
   struct f_val values[];
 };
 
-void rt_notify_aggregated(struct channel *c, struct network *net, struct rte *new_changed, struct rte *old_changed,
-                          struct rte *new_best, struct rte *old_best, int refeed);
-
 rte *export_filter(struct channel *c, struct rte *rt0, struct rte **rt_free, int silent);
 
 /*

nest/rt-attr.c

@@ -75,6 +75,8 @@ const char * const rta_src_names[RTS_MAX] = {
   [RTS_PIPE]		= "pipe",
   [RTS_BABEL]		= "Babel",
   [RTS_RPKI]		= "RPKI",
+  [RTS_PERF]		= "Perf",
+  [RTS_AGGREGATED]	= "aggregated",
 };
 
 const char * rta_dest_names[RTD_MAX] = {
@@ -1272,7 +1274,8 @@ rta_dump(rta *a)
   static char *rts[] = { "", "RTS_STATIC", "RTS_INHERIT", "RTS_DEVICE",
 			 "RTS_STAT_DEV", "RTS_REDIR", "RTS_RIP",
 			 "RTS_OSPF", "RTS_OSPF_IA", "RTS_OSPF_EXT1",
-			 "RTS_OSPF_EXT2", "RTS_BGP", "RTS_PIPE", "RTS_BABEL" };
+			 "RTS_OSPF_EXT2", "RTS_BGP", "RTS_PIPE", "RTS_BABEL",
+			 "RTS_RPKI", "RTS_PERF", "RTS_AGGREGATED", };
   static char *rtd[] = { "", " DEV", " HOLE", " UNREACH", " PROHIBIT" };
 
   debug("pref=%d uc=%d %s %s%s h=%04x",

nest/rt-show.c

@@ -150,21 +150,12 @@ rt_show_net(struct cli *c, net *n, struct rt_show_data *d)
 	  if (!e)
 	  { e = ee; goto skip; }
 	}
-      else if ((d->export_mode == RSEM_EXPORT) && (ec->ra_mode == RA_AGGREGATED))
-    {
-      rte *rte_free;
-      e = rt_export_aggregated(ec, n, &rte_free, c->show_pool, 1);
-      pass = 1;
-
-	  if (!e)
-	  { e = ee; goto skip; }
-    }
       else if (d->export_mode)
 	{
 	  struct proto *ep = ec->proto;
 	  int ic = ep->preexport ? ep->preexport(ec, e) : 0;
 
-	  if (ec->ra_mode == RA_OPTIMAL || ec->ra_mode == RA_MERGED || ec->ra_mode == RA_AGGREGATED)
+	  if (ec->ra_mode == RA_OPTIMAL || ec->ra_mode == RA_MERGED)
 	    pass = 1;
 
 	  if (ic < 0)

nest/rt-table.c

@@ -122,7 +122,7 @@
 pool *rt_table_pool;
 
 static slab *rte_slab;
-static linpool *rte_update_pool;
+linpool *rte_update_pool;
 
 list routing_tables;
 
@@ -1190,10 +1190,6 @@ rte_announce(rtable *tab, uint type, net *net, rte *new, rte *old,
     case RA_MERGED:
       rt_notify_merged(c, net, new, old, new_best, old_best, 0);
       break;
-
-    case RA_AGGREGATED:
-      rt_notify_aggregated(c, net, new, old, new_best, old_best, 0);
-      break;
     }
   }
 }
@@ -1265,7 +1261,7 @@ rte_free_quick(rte *e)
   sl_free(e);
 }
 
-static int
+int
 rte_same(rte *x, rte *y)
 {
   /* rte.flags / rte.pflags are not checked, as they are internal to rtable */
@@ -3028,8 +3024,6 @@ do_feed_channel(struct channel *c, net *n, rte *e)
     rt_notify_accepted(c, n, NULL, NULL, c->refeeding);
   else if (c->ra_mode == RA_MERGED)
     rt_notify_merged(c, n, NULL, NULL, e, e, c->refeeding);
-  else if (c->ra_mode == RA_AGGREGATED)
-    rt_notify_aggregated(c, n, NULL, NULL, e, e, c->refeeding);
   else /* RA_BASIC */
     rt_notify_basic(c, n, e, e, c->refeeding);
   rte_update_unlock();
@@ -3069,8 +3063,7 @@ rt_feed_channel(struct channel *c)
 
       if ((c->ra_mode == RA_OPTIMAL) ||
 	  (c->ra_mode == RA_ACCEPTED) ||
-	  (c->ra_mode == RA_MERGED)   ||
+	  (c->ra_mode == RA_MERGED))
-      (c->ra_mode == RA_AGGREGATED))
 	if (rte_is_valid(e))
 	  {
 	    /* In the meantime, the protocol may fell down */

proto/aggregator/Makefile

@@ -0,0 +1,6 @@
+src := aggregator.c
+obj := $(src-o-files)
+$(all-daemon)
+$(cf-local)
+
+tests_objs := $(tests_objs) $(src-o-files)

proto/aggregator/aggregator.c

@@ -0,0 +1,730 @@
+/*
+ *	BIRD Internet Routing Daemon -- Route aggregation
+ *
+ *	(c) 2023--2023 Igor Putovny <igor.putovny@nic.cz>
+ *	(c) 2023       CZ.NIC, z.s.p.o.
+ *
+ *	Can be freely distributed and used under the terms of the GNU GPL.
+ */
+
+/**
+ * DOC: Route aggregation
+ *
+ * This is an implementation of route aggregation functionality.
+ * It enables user to specify a set of route attributes in the configuarion file
+ * and then, for a given destination (net), aggregate routes with the same
+ * values of these attributes into a single multi-path route.
+ *
+ * Structure &channel contains pointer to aggregation list which is represented
+ * by &aggr_list_linearized. In rt_notify_aggregated(), attributes from this
+ * list are evaluated for every route of a given net and results are stored
+ * in &rte_val_list which contains pointer to this route and array of &f_val.
+ * Array of pointers to &rte_val_list entries is sorted using
+ * sort_rte_val_list(). For comparison of &f_val structures, val_compare()
+ * is used. Comparator function is written so that sorting is stable. If all
+ * attributes have the same values, routes are compared by their global IDs.
+ *
+ * After sorting, &rte_val_list entries containing equivalent routes will be
+ * adjacent to each other. Function process_rte_list() iterates through these
+ * entries to identify sequences of equivalent routes. New route will be
+ * created for each such sequence, even if only from a single route.
+ * Only attributes from the aggreagation list will be set for the new route.
+ * New &rta is created and prepare_rta() is used to copy static and dynamic
+ * attributes to new &rta from &rta of the original route. New route is created
+ * by create_merged_rte() from new &rta and exported to the routing table.
+ */
+
+#undef LOCAL_DEBUG
+
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
+#endif
+
+#include "nest/bird.h"
+#include "nest/iface.h"
+#include "filter/filter.h"
+#include "proto/aggregator/aggregator.h"
+
+#include <stdlib.h>
+/*
+#include "nest/route.h"
+#include "nest/iface.h"
+#include "lib/resource.h"
+#include "lib/event.h"
+#include "lib/timer.h"
+#include "lib/string.h"
+#include "conf/conf.h"
+#include "filter/filter.h"
+#include "filter/data.h"
+#include "lib/hash.h"
+#include "lib/string.h"
+#include "lib/alloca.h"
+#include "lib/flowspec.h"
+*/
+
+/* Context of &f_val comparison. */
+struct cmp_ctx {
+  const struct aggregator_proto *p;
+  const struct network *net;
+  const int val_count;
+  u32 failed:1;
+};
+
+extern linpool *rte_update_pool;
+
+/*
+ * Set static attribute in @rta from static attribute in @old according to @sa.
+ */
+static void
+rta_set_static_attr(struct rta *rta, const struct rta *old, struct f_static_attr sa)
+{
+  switch (sa.sa_code)
+  {
+    case SA_NET:
+      break;
+
+    case SA_FROM:
+      rta->from = old->from;
+      break;
+
+    case SA_GW:
+      rta->dest = RTD_UNICAST;
+      rta->nh.gw = old->nh.gw;
+      rta->nh.iface = old->nh.iface;
+      rta->nh.next = NULL;
+      rta->hostentry = NULL;
+      rta->nh.labels = 0;
+      break;
+
+    case SA_SCOPE:
+      rta->scope = old->scope;
+      break;
+
+    case SA_DEST:
+      rta->dest = old->dest;
+      rta->nh.gw = IPA_NONE;
+      rta->nh.iface = NULL;
+      rta->nh.next = NULL;
+      rta->hostentry = NULL;
+      rta->nh.labels = 0;
+      break;
+
+    case SA_IFNAME:
+      rta->dest = RTD_UNICAST;
+      rta->nh.gw = IPA_NONE;
+      rta->nh.iface = old->nh.iface;
+      rta->nh.next = NULL;
+      rta->hostentry = NULL;
+      rta->nh.labels = 0;
+      break;
+
+    case SA_GW_MPLS:
+      rta->nh.labels = old->nh.labels;
+      memcpy(&rta->nh.label, &old->nh.label, sizeof(u32) * old->nh.labels);
+      break;
+
+    case SA_WEIGHT:
+      rta->nh.weight = old->nh.weight;
+      break;
+
+    case SA_PREF:
+      rta->pref = old->pref;
+      break;
+
+    default:
+      bug("Invalid static attribute access (%u/%u)", sa.f_type, sa.sa_code);
+  }
+}
+
+/*
+ * Compare list of &f_val entries.
+ * @count: number of &f_val entries
+ */
+static int
+same_val_list(const struct f_val *v1, const struct f_val *v2, uint len)
+{
+  for (uint i = 0; i < len; i++)
+    if (!val_same(&v1[i], &v2[i]))
+      return 0;
+
+  return 1;
+}
+
+/*
+ * Create and export new merged route.
+ * @old: first route in a sequence of equivalent routes that are to be merged
+ * @rte_val: first element in a sequence of equivalent rte_val_list entries
+ * @length: number of equivalent routes that are to be merged (at least 1)
+ * @ail: aggregation list
+ */
+static void
+aggregator_bucket_update(struct aggregator_proto *p, struct aggregator_bucket *bucket, struct network *net)
+{
+  /* Empty bucket */
+  if (!bucket->rte)
+  {
+    rte_update2(p->dst, net->n.addr, NULL, bucket->last_src);
+    bucket->last_src = NULL;
+    return;
+  }
+
+  /* Allocate RTA and EA list */
+  struct rta *rta = allocz(rta_size(bucket->rte->attrs));
+  rta->dest = RTD_UNREACHABLE;
+  rta->source = RTS_AGGREGATED;
+  rta->scope = SCOPE_UNIVERSE;
+
+  struct ea_list *eal = allocz(sizeof(struct ea_list) + sizeof(struct eattr) * p->aggr_on_da_count);
+  eal->next = NULL;
+  eal->count = 0;
+  rta->eattrs = eal;
+
+  /* Seed the attributes from aggregator rule */
+  for (uint i = 0; i < p->aggr_on_count; i++)
+  {
+    if (p->aggr_on[i].type == AGGR_ITEM_DYNAMIC_ATTR)
+    {
+      u32 ea_code = p->aggr_on[i].da.ea_code;
+      const struct eattr *e = ea_find(bucket->rte->attrs->eattrs, ea_code);
+
+      if (e)
+        eal->attrs[eal->count++] = *e;
+    }
+    else if (p->aggr_on[i].type == AGGR_ITEM_STATIC_ATTR)
+      rta_set_static_attr(rta, bucket->rte->attrs, p->aggr_on[i].sa);
+  }
+
+  struct rte *new = rte_get_temp(rta, bucket->rte->src);
+  new->net = net;
+
+  log("=============== CREATE MERGED ROUTE ===============");
+  log("New route created: id = %d, protocol: %s", new->src->global_id, new->src->proto->name);
+  log("===================================================");
+
+
+  /* merge filter needs one argument called "routes" */
+  struct f_val val = {
+    .type = T_ROUTES_BLOCK,
+    .val.rte = bucket->rte,
+  };
+
+  /* Actually run the filter */
+  enum filter_return fret = f_eval_rte(p->merge_by, &new, rte_update_pool, 1, &val, 0);
+
+  /* Src must be stored now, rte_update2() may return new */
+  struct rte_src *new_src = new ? new->src : NULL;
+
+  /* Finally import the route */
+  switch (fret)
+  {
+    /* Pass the route to the protocol */
+    case F_ACCEPT:
+      rte_update2(p->dst, net->n.addr, new, bucket->last_src ?: new->src);
+      break;
+
+    /* Something bad happened */
+    default:
+      ASSERT_DIE(fret == F_ERROR);
+      /* fall through */
+
+    /* We actually don't want this route */
+    case F_REJECT:
+      if (bucket->last_src)
+	rte_update2(p->dst, net->n.addr, NULL, bucket->last_src);
+      break;
+  }
+
+  /* Switch source lock for bucket->last_src */
+  if (bucket->last_src != new_src)
+  {
+    if (new_src)
+      rt_lock_source(new_src);
+    if (bucket->last_src)
+      rt_unlock_source(bucket->last_src);
+
+    bucket->last_src = new_src;
+  }
+}
+
+/*
+ * Reload all the buckets on reconfiguration if merge filter has changed.
+ * TODO: make this splitted
+ */
+static void
+aggregator_reload_buckets(void *data)
+{
+  struct aggregator_proto *p = data;
+
+  HASH_WALK(p->buckets, next_hash, b)
+    if (b->rte)
+      aggregator_bucket_update(p, b, b->rte->net);
+  HASH_WALK_END;
+}
+
+
+/*
+ * Evaluate static attribute of @rt1 according to @sa
+ * and store result in @pos.
+ */
+static void
+eval_static_attr(const struct rte *rt1, struct f_static_attr sa, struct f_val *pos)
+{
+  const struct rta *rta = rt1->attrs;
+
+#define RESULT(_type, value, result)    \
+  do {                                  \
+    pos->type = _type;                  \
+    pos->val.value = result;            \
+  } while (0)
+
+  switch (sa.sa_code)
+  {
+    case SA_NET:	RESULT(sa.f_type, net, rt1->net->n.addr); break;
+    case SA_FROM:       RESULT(sa.f_type, ip, rta->from); break;
+    case SA_GW:	        RESULT(sa.f_type, ip, rta->nh.gw); break;
+    case SA_PROTO:	    RESULT(sa.f_type, s, rt1->src->proto->name); break;
+    case SA_SOURCE:	    RESULT(sa.f_type, i, rta->source); break;
+    case SA_SCOPE:	    RESULT(sa.f_type, i, rta->scope); break;
+    case SA_DEST:	    RESULT(sa.f_type, i, rta->dest); break;
+    case SA_IFNAME:	    RESULT(sa.f_type, s, rta->nh.iface ? rta->nh.iface->name : ""); break;
+    case SA_IFINDEX:	RESULT(sa.f_type, i, rta->nh.iface ? rta->nh.iface->index : 0); break;
+    case SA_WEIGHT:	    RESULT(sa.f_type, i, rta->nh.weight + 1); break;
+    case SA_PREF:	    RESULT(sa.f_type, i, rta->pref); break;
+    case SA_GW_MPLS:    RESULT(sa.f_type, i, rta->nh.labels ? rta->nh.label[0] : MPLS_NULL); break;
+    default:
+      bug("Invalid static attribute access (%u/%u)", sa.f_type, sa.sa_code);
+  }
+
+#undef RESULT
+}
+
+/*
+ * Evaluate dynamic attribute of @rt1 according to @da
+ * and store result in @pos.
+ */
+static void
+eval_dynamic_attr(const struct rte *rt1, struct f_dynamic_attr da, struct f_val *pos)
+{
+  const struct rta *rta = rt1->attrs;
+  const struct eattr *e = ea_find(rta->eattrs, da.ea_code);
+
+#define RESULT(_type, value, result)    \
+  do {                                  \
+    pos->type = _type;                  \
+    pos->val.value = result;            \
+  } while (0)
+
+#define RESULT_VOID         \
+  do {                      \
+    pos->type = T_VOID;     \
+  } while (0)
+
+  if (!e)
+  {
+    /* A special case: undefined as_path looks like empty as_path */
+    if (da.type == EAF_TYPE_AS_PATH)
+    {
+      RESULT(T_PATH, ad, &null_adata);
+      return;
+    }
+
+    /* The same special case for int_set */
+    if (da.type == EAF_TYPE_INT_SET)
+    {
+      RESULT(T_CLIST, ad, &null_adata);
+      return;
+    }
+
+    /* The same special case for ec_set */
+    if (da.type == EAF_TYPE_EC_SET)
+    {
+      RESULT(T_ECLIST, ad, &null_adata);
+      return;
+    }
+
+    /* The same special case for lc_set */
+    if (da.type == EAF_TYPE_LC_SET)
+    {
+      RESULT(T_LCLIST, ad, &null_adata);
+      return;
+    }
+
+    /* Undefined value */
+    RESULT_VOID;
+    return;
+  }
+
+  switch (e->type & EAF_TYPE_MASK)
+  {
+    case EAF_TYPE_INT:
+      RESULT(da.f_type, i, e->u.data);
+      break;
+    case EAF_TYPE_ROUTER_ID:
+      RESULT(T_QUAD, i, e->u.data);
+      break;
+    case EAF_TYPE_OPAQUE:
+      RESULT(T_ENUM_EMPTY, i, 0);
+      break;
+    case EAF_TYPE_IP_ADDRESS:
+      RESULT(T_IP, ip, *((ip_addr *) e->u.ptr->data));
+      break;
+    case EAF_TYPE_AS_PATH:
+      RESULT(T_PATH, ad, e->u.ptr);
+      break;
+    case EAF_TYPE_BITFIELD:
+      RESULT(T_BOOL, i, !!(e->u.data & (1u << da.bit)));
+      break;
+    case EAF_TYPE_INT_SET:
+      RESULT(T_CLIST, ad, e->u.ptr);
+      break;
+    case EAF_TYPE_EC_SET:
+      RESULT(T_ECLIST, ad, e->u.ptr);
+      break;
+    case EAF_TYPE_LC_SET:
+      RESULT(T_LCLIST, ad, e->u.ptr);
+      break;
+    default:
+      bug("Unknown dynamic attribute type");
+  }
+
+#undef RESULT
+#undef RESULT_VOID
+}
+
+static inline u32 aggr_route_hash(const rte *e)
+{
+  struct {
+    net *net;
+    struct rte_src *src;
+  } obj = {
+    .net = e->net,
+    .src = e->src,
+  };
+
+  return mem_hash(&obj, sizeof obj);
+}
+
+#define AGGR_RTE_KEY(n)			(&(n)->rte)
+#define AGGR_RTE_NEXT(n)		((n)->next_hash)
+#define AGGR_RTE_EQ(a,b)		(((a)->src == (b)->src) && ((a)->net == (b)->net))
+#define AGGR_RTE_FN(_n)			aggr_route_hash(_n)
+#define AGGR_RTE_ORDER			4 /* Initial */
+
+#define AGGR_RTE_REHASH			aggr_rte_rehash
+#define AGGR_RTE_PARAMS			/8, *2, 2, 2, 4, 24
+
+HASH_DEFINE_REHASH_FN(AGGR_RTE, struct aggregator_route);
+
+
+#define AGGR_BUCK_KEY(n)		(n)
+#define AGGR_BUCK_NEXT(n)		((n)->next_hash)
+#define AGGR_BUCK_EQ(a,b)		(((a)->hash == (b)->hash) && (same_val_list((a)->aggr_data, (b)->aggr_data, p->aggr_on_count)))
+#define AGGR_BUCK_FN(n)			((n)->hash)
+#define AGGR_BUCK_ORDER			4 /* Initial */
+
+#define AGGR_BUCK_REHASH		aggr_buck_rehash
+#define AGGR_BUCK_PARAMS		/8, *2, 2, 2, 4, 24
+
+HASH_DEFINE_REHASH_FN(AGGR_BUCK, struct aggregator_bucket);
+
+
+#define AGGR_DATA_MEMSIZE	(sizeof(struct f_val) * p->aggr_on_count)
+
+static void
+aggregator_rt_notify(struct proto *P, struct channel *src_ch, net *net, rte *new, rte *old)
+{
+  struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P);
+  ASSERT_DIE(src_ch == p->src);
+  struct aggregator_bucket *new_bucket = NULL, *old_bucket = NULL;
+  struct aggregator_route *old_route = NULL;
+
+  /* Find the objects for the old route */
+  if (old)
+    old_route = HASH_FIND(p->routes, AGGR_RTE, old);
+
+  if (old_route)
+    old_bucket = old_route->bucket;
+
+  /* Find the bucket for the new route */
+  if (new)
+  {
+    /* Routes are identical, do nothing */
+    if (old_route && rte_same(&old_route->rte, new))
+      return;
+
+    /* Evaluate route attributes. */
+    struct aggregator_bucket *tmp_bucket = sl_allocz(p->bucket_slab);
+
+    for (uint val_idx = 0; val_idx < p->aggr_on_count; val_idx++)
+    {
+      int type = p->aggr_on[val_idx].type;
+
+      switch (type)
+      {
+        case AGGR_ITEM_TERM: {
+          const struct f_line *line = p->aggr_on[val_idx].line;
+          struct rte *rt1 = new;
+          enum filter_return fret = f_eval_rte(line, &new, rte_update_pool, 0, NULL, &tmp_bucket->aggr_data[val_idx]);
+
+          if (rt1 != new)
+          {
+            rte_free(rt1);
+            log(L_WARN "Aggregator rule modifies the route, reverting");
+          }
+
+          if (fret > F_RETURN)
+            log(L_WARN "%s.%s: Wrong number of items left on stack after evaluation of aggregation list", rt1->src->proto->name, rt1->sender);
+
+          break;
+        }
+
+        case AGGR_ITEM_STATIC_ATTR: {
+          struct f_val *pos = &tmp_bucket->aggr_data[val_idx];
+          eval_static_attr(new, p->aggr_on[val_idx].sa, pos);
+          break;
+        }
+
+        case AGGR_ITEM_DYNAMIC_ATTR: {
+          struct f_val *pos = &tmp_bucket->aggr_data[val_idx];
+          eval_dynamic_attr(new, p->aggr_on[val_idx].da, pos);
+          break;
+        }
+
+        default:
+          break;
+      }
+    }
+
+    /* Compute the hash */
+    tmp_bucket->hash = mem_hash(tmp_bucket->aggr_data, AGGR_DATA_MEMSIZE);
+
+    /* Find the existing bucket */
+    if (new_bucket = HASH_FIND(p->buckets, AGGR_BUCK, tmp_bucket))
+      sl_free(tmp_bucket);
+    else
+    {
+      new_bucket = tmp_bucket;
+      HASH_INSERT2(p->buckets, AGGR_BUCK, p->p.pool, new_bucket);
+    }
+
+    /* Store the route attributes */
+    if (rta_is_cached(new->attrs))
+      rta_clone(new->attrs);
+    else
+      new->attrs = rta_lookup(new->attrs);
+
+    /* Insert the new route into the bucket */
+    struct aggregator_route *arte = sl_alloc(p->route_slab);
+    *arte = (struct aggregator_route) {
+      .bucket = new_bucket,
+      .rte = *new,
+    };
+    arte->rte.next = new_bucket->rte,
+    new_bucket->rte = &arte->rte;
+    new_bucket->count++;
+    HASH_INSERT2(p->routes, AGGR_RTE, p->p.pool, arte);
+  }
+
+  /* Remove the old route from its bucket */
+  if (old_bucket)
+  {
+    for (struct rte **k = &old_bucket->rte; *k; k = &(*k)->next)
+      if (*k == &old_route->rte)
+      {
+	*k = (*k)->next;
+	break;
+      }
+
+    old_bucket->count--;
+    HASH_REMOVE2(p->routes, AGGR_RTE, p->p.pool, old_route);
+    rta_free(old_route->rte.attrs);
+    sl_free(old_route);
+  }
+
+  /* Announce changes */
+  if (old_bucket)
+    aggregator_bucket_update(p, old_bucket, net);
+
+  if (new_bucket && (new_bucket != old_bucket))
+    aggregator_bucket_update(p, new_bucket, net);
+
+  /* Cleanup the old bucket if empty */
+  if (old_bucket && (!old_bucket->rte || !old_bucket->count))
+  {
+    ASSERT_DIE(!old_bucket->rte && !old_bucket->count);
+    HASH_REMOVE2(p->buckets, AGGR_BUCK, p->p.pool, old_bucket);
+    sl_free(old_bucket);
+  }
+}
+
+static int
+aggregator_preexport(struct channel *C, struct rte *new)
+{
+  struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, C->proto);
+  /* Reject our own routes */
+  if (new->sender == p->dst)
+    return -1;
+
+  /* Disallow aggregating already aggregated routes */
+  if (new->attrs->source == RTS_AGGREGATED)
+  {
+    log(L_ERR "Multiple aggregations of the same route not supported in BIRD 2.");
+    return -1;
+  }
+
+  return 0;
+}
+
+static void
+aggregator_postconfig(struct proto_config *CF)
+{
+  struct aggregator_config *cf = SKIP_BACK(struct aggregator_config, c, CF);
+
+  if (!cf->dst->table)
+    cf_error("Source table not specified");
+
+  if (!cf->src->table)
+    cf_error("Destination table not specified");
+
+  if (cf->dst->table->addr_type != cf->src->table->addr_type)
+    cf_error("Both tables must be of the same type");
+
+  cf->dst->in_filter = cf->src->in_filter;
+
+  cf->src->in_filter = FILTER_REJECT;
+  cf->dst->out_filter = FILTER_REJECT;
+
+  cf->dst->debug = cf->src->debug;
+}
+
+static struct proto *
+aggregator_init(struct proto_config *CF)
+{
+  struct proto *P = proto_new(CF);
+  struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P);
+  struct aggregator_config *cf = SKIP_BACK(struct aggregator_config, c, CF);
+
+  proto_configure_channel(P, &p->src, cf->src);
+  proto_configure_channel(P, &p->dst, cf->dst);
+
+  p->aggr_on_count = cf->aggr_on_count;
+  p->aggr_on_da_count = cf->aggr_on_da_count;
+  p->aggr_on = cf->aggr_on;
+  p->merge_by = cf->merge_by;
+
+  P->rt_notify = aggregator_rt_notify;
+  P->preexport = aggregator_preexport;
+
+  return P;
+}
+
+static int
+aggregator_start(struct proto *P)
+{
+  struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P);
+
+  p->bucket_slab = sl_new(P->pool, sizeof(struct aggregator_bucket) + AGGR_DATA_MEMSIZE);
+  HASH_INIT(p->buckets, P->pool, AGGR_BUCK_ORDER);
+
+  p->route_slab = sl_new(P->pool, sizeof(struct aggregator_route));
+  HASH_INIT(p->routes, P->pool, AGGR_RTE_ORDER);
+
+  p->reload_buckets = (event) {
+    .hook = aggregator_reload_buckets,
+    .data = p,
+  };
+
+  return PS_UP;
+}
+
+static int
+aggregator_shutdown(struct proto *P)
+{
+  struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P);
+
+  HASH_WALK_DELSAFE(p->buckets, next_hash, b)
+  {
+    while (b->rte)
+    {
+      struct aggregator_route *arte = SKIP_BACK(struct aggregator_route, rte, b->rte);
+      b->rte = arte->rte.next;
+      b->count--;
+      HASH_REMOVE(p->routes, AGGR_RTE, arte);
+      rta_free(arte->rte.attrs);
+      sl_free(arte);
+    }
+
+    ASSERT_DIE(b->count == 0);
+    HASH_REMOVE(p->buckets, AGGR_BUCK, b);
+    sl_free(b);
+  }
+  HASH_WALK_END;
+
+  return PS_DOWN;
+}
+
+static int
+aggregator_reconfigure(struct proto *P, struct proto_config *CF)
+{
+  struct aggregator_proto *p = SKIP_BACK(struct aggregator_proto, p, P);
+  struct aggregator_config *cf = SKIP_BACK(struct aggregator_config, c, CF);
+
+  TRACE(D_EVENTS, "Reconfiguring");
+
+  /* Compare numeric values (shortcut) */
+  if (cf->aggr_on_count != p->aggr_on_count)
+    return 0;
+
+  if (cf->aggr_on_da_count != p->aggr_on_da_count)
+    return 0;
+
+  /* Compare aggregator rule */
+  for (uint i = 0; i < p->aggr_on_count; i++)
+    switch (cf->aggr_on[i].type)
+    {
+      case AGGR_ITEM_TERM:
+	if (!f_same(cf->aggr_on[i].line, p->aggr_on[i].line))
+	  return 0;
+	break;
+      case AGGR_ITEM_STATIC_ATTR:
+	if (memcmp(&cf->aggr_on[i].sa, &p->aggr_on[i].sa, sizeof(struct f_static_attr)) != 0)
+	  return 0;
+	break;
+      case AGGR_ITEM_DYNAMIC_ATTR:
+	if (memcmp(&cf->aggr_on[i].da, &p->aggr_on[i].da, sizeof(struct f_dynamic_attr)) != 0)
+	  return 0;
+	break;
+      default:
+	bug("Broken aggregator rule");
+    }
+
+  /* Compare merge filter */
+  if (!f_same(cf->merge_by, p->merge_by))
+    ev_schedule(&p->reload_buckets);
+
+  p->aggr_on = cf->aggr_on;
+  p->merge_by = cf->merge_by;
+
+  return 1;
+}
+
+struct protocol proto_aggregator = {
+  .name =		"Aggregator",
+  .template =		"aggregator%d",
+  .class =		PROTOCOL_AGGREGATOR,
+  .preference =		1,
+  .channel_mask =	NB_ANY,
+  .proto_size =		sizeof(struct aggregator_proto),
+  .config_size =	sizeof(struct aggregator_config),
+  .postconfig =		aggregator_postconfig,
+  .init =		aggregator_init,
+  .start =		aggregator_start,
+  .shutdown =		aggregator_shutdown,
+  .reconfigure =	aggregator_reconfigure,
+};
+
+void
+aggregator_build(void)
+{
+  proto_build(&proto_aggregator);
+}

proto/aggregator/aggregator.h

@@ -0,0 +1,86 @@
+/*
+ *	BIRD -- Aggregator Pseudoprotocol
+ *
+ *	(c) 2023       Igor Putovny <igor.putovny@nic.cz>
+ *	(c) 2023       Maria Matejka <mq@ucw.cz>
+ *	(c) 2023       CZ.NIC z.s.p.o.
+ *
+ *	Can be freely distributed and used under the terms of the GNU GPL.
+ *
+ *	This file contains the data structures used by Babel.
+ */
+
+#ifndef _BIRD_AGGREGATOR_H_
+#define _BIRD_AGGREGATOR_H_
+
+#include "nest/bird.h"
+#include "nest/protocol.h"
+#include "lib/hash.h"
+
+struct aggregator_config {
+  struct proto_config c;
+  struct channel_config *src, *dst;
+  uint aggr_on_count;
+  uint aggr_on_da_count;
+  struct aggr_item *aggr_on;
+  const struct f_line *merge_by;
+};
+
+struct aggregator_route {
+  struct aggregator_route *next_hash;
+  struct aggregator_bucket *bucket;
+  struct rte rte;
+};
+
+struct aggregator_bucket {
+  struct aggregator_bucket *next_hash;
+  struct rte *rte;			/* Pointer to struct aggregator_route.rte */
+  struct rte_src *last_src;		/* Which src we announced the bucket last with */
+  u32 count;
+  u32 hash;
+  struct f_val aggr_data[0];
+};
+
+struct aggregator_proto {
+  struct proto p;
+  struct channel *src, *dst;
+
+  /* Buckets by aggregator rule */
+  HASH(struct aggregator_bucket) buckets;
+  slab *bucket_slab;
+
+  /* Routes by net and src */
+  HASH(struct aggregator_route) routes;
+  slab *route_slab;
+
+  /* Aggregator rule */
+  uint aggr_on_count;
+  uint aggr_on_da_count;
+  struct aggr_item *aggr_on;
+
+  /* Merge filter */
+  const struct f_line *merge_by;
+  event reload_buckets;
+};
+
+enum aggr_item_type {
+  AGGR_ITEM_TERM,
+  AGGR_ITEM_STATIC_ATTR,
+  AGGR_ITEM_DYNAMIC_ATTR,
+};
+
+struct aggr_item {
+  enum aggr_item_type type;
+  union {
+    struct f_static_attr sa;
+    struct f_dynamic_attr da;
+    const struct f_line *line;
+  };
+};
+
+struct aggr_item_node {
+  const struct aggr_item_node *next;
+  struct aggr_item i;
+};
+
+#endif

proto/aggregator/config.Y

@@ -0,0 +1,136 @@
+/*
+ *	BIRD -- Aggregator configuration
+ *
+ *	(c) 2023       Igor Putovny <igor.putovny@nic.cz>
+ *	(c) 2023       Maria Matejka <mq@ucw.cz>
+ *	(c) 2023       CZ.NIC z.s.p.o.
+ *
+ *	Can be freely distributed and used under the terms of the GNU GPL.
+ */
+
+CF_HDR
+
+#include "proto/aggregator/aggregator.h"
+
+CF_DEFINES
+
+#define AGGREGATOR_CFG ((struct aggregator_config *) this_proto)
+#define AGGR_ITEM_ALLOC ((struct aggr_item_node *) cfg_allocz(sizeof(struct aggr_item_node)))
+
+
+CF_DECLS
+
+CF_KEYWORDS(AGGREGATOR, AGGREGATE, ON, MERGE, BY)
+
+%type <ai> aggr_item aggr_list
+
+CF_GRAMMAR
+
+toksym: AGGREGATOR | AGGREGATE | MERGE | BY ;
+
+proto: aggregator_proto ;
+
+aggregator_proto_start: proto_start AGGREGATOR
+{
+  this_proto = proto_config_new(&proto_aggregator, $1);
+  this_channel = AGGREGATOR_CFG->src = channel_config_new(NULL, "source", 0, this_proto);
+  AGGREGATOR_CFG->dst = channel_config_new(NULL, "destination", 0, this_proto);
+
+  AGGREGATOR_CFG->src->ra_mode = AGGREGATOR_CFG->dst->ra_mode = RA_ANY;
+};
+
+aggregator_proto_item:
+   proto_item
+ | channel_item_
+ | PEER TABLE rtable { AGGREGATOR_CFG->dst->table = $3; }
+ | AGGREGATE ON aggr_list {
+    if (AGGREGATOR_CFG->aggr_on)
+      cf_error("Only one aggregate on clause allowed");
+
+    _Bool net_present = 0;
+    int count = 0;
+
+    for (const struct aggr_item_node *item = $3; item; item = item->next) {
+      log(L_WARN "type %d sacode %d", item->i.type, item->i.sa.sa_code);
+      if (item->i.type == AGGR_ITEM_STATIC_ATTR && item->i.sa.sa_code == SA_NET)
+	net_present = 1;
+
+      count++;
+    }
+
+   if (!net_present)
+     cf_error("'NET' must be present");
+
+   AGGREGATOR_CFG->aggr_on = cfg_alloc(sizeof(struct aggr_item) * count);
+
+   int pos = 0;
+   for (const struct aggr_item_node *item = $3; item; item = item->next) {
+     if (item->i.type == AGGR_ITEM_DYNAMIC_ATTR)
+       AGGREGATOR_CFG->aggr_on_da_count++;
+
+     AGGREGATOR_CFG->aggr_on[pos++] = item->i;
+   }
+
+   AGGREGATOR_CFG->aggr_on_count = pos;
+ }
+ | MERGE BY {
+   cf_push_block_scope(new_config);
+   cf_create_symbol(new_config, "routes", SYM_VARIABLE | T_ROUTES_BLOCK, offset, f_new_var(sym_->scope));
+ } function_body {
+   cf_pop_block_scope(new_config);
+   $4->args++;
+   AGGREGATOR_CFG->merge_by = $4;
+ }
+;
+
+aggregator_proto_opts: /* empty */ | aggregator_proto_opts aggregator_proto_item ';' ;
+aggregator_proto: aggregator_proto_start proto_name '{' aggregator_proto_opts '}' ;
+
+
+aggr_list:
+   aggr_item
+ | aggr_list ',' aggr_item {
+       if ($3 == NULL) {
+         $$ = $1;
+       } else {
+         $$ = $3;
+         $$->next = $1;
+       }
+   }
+ ;
+
+aggr_item:
+   '(' term ')' {
+       $$ = AGGR_ITEM_ALLOC;
+       $$->i.type = AGGR_ITEM_TERM;
+       $$->i.line = f_linearize($2, 1);
+    }
+  | CF_SYM_KNOWN {
+      switch ($1->class) {
+        case SYM_ATTRIBUTE:
+          $$ = AGGR_ITEM_ALLOC;
+          $$->i.type = AGGR_ITEM_DYNAMIC_ATTR;
+          $$->i.da = *$1->attribute;
+          break;
+        case SYM_CONSTANT_RANGE:
+          $$ = NULL;
+          break;
+        default:
+          cf_error("Can't aggregate on symbol type %s.", cf_symbol_class_name($1));
+      }
+    }
+  | dynamic_attr {
+      $$ = AGGR_ITEM_ALLOC;
+      $$->i.type = AGGR_ITEM_DYNAMIC_ATTR;
+      $$->i.da = $1;
+    }
+  | static_attr {
+      $$ = AGGR_ITEM_ALLOC;
+      $$->i.type = AGGR_ITEM_STATIC_ATTR;
+      $$->i.sa = $1;
+    }
+  ;
+
+CF_CODE
+
+CF_END

proto/pipe/config.Y

@@ -9,7 +9,6 @@
 CF_HDR
 
 #include "proto/pipe/pipe.h"
-#include "nest/route.h"
 
 CF_DEFINES
 
@@ -17,7 +16,7 @@ CF_DEFINES
 
 CF_DECLS
 
-CF_KEYWORDS(PIPE, PEER, TABLE, AGGREGATED, IMPORT, EXPORT, AGGREGATE)
+CF_KEYWORDS(PIPE, PEER, TABLE)
 
 CF_GRAMMAR
 
@@ -42,8 +41,6 @@ pipe_proto:
  | pipe_proto proto_item ';'
  | pipe_proto channel_item_ ';'
  | pipe_proto PEER TABLE rtable ';' { PIPE_CFG->peer = $4; }
- | pipe_proto MERGE PATHS IMPORT bool kern_mp_limit { PIPE_CFG->config_import.limit = $5 ? $6 : 0; PIPE_CFG->config_import.use_aggregator = 0; }
- | pipe_proto MERGE PATHS EXPORT bool kern_mp_limit { PIPE_CFG->config_export.limit = $5 ? $6 : 0; PIPE_CFG->config_export.use_aggregator = 0; }
  ;
 
 CF_CODE

proto/pipe/pipe.c

@@ -1,4 +1,4 @@
-    /*
+/*
  *	BIRD -- Table-to-Table Routing Protocol a.k.a Pipe
  *
  *	(c) 1999--2000 Martin Mares <mj@ucw.cz>
@@ -81,7 +81,7 @@ pipe_rt_notify(struct proto *P, struct channel *src_ch, net *n, rte *new, rte *o
     }
 
   src_ch->table->pipe_busy = 1;
-  rte_update2(dst, n->n.addr, e, old ? old->src : new->src);
+  rte_update2(dst, n->n.addr, e, src);
   src_ch->table->pipe_busy = 0;
 }
 
@@ -138,16 +138,13 @@ pipe_configure_channels(struct pipe_proto *p, struct pipe_config *cf)
 {
   struct channel_config *cc = proto_cf_main_channel(&cf->c);
 
-  log("pipe configure channels: ra_mode: %d", cc->ra_mode);
-  log("pipe configure channels: ai_aggr: %p", cc->ai_aggr);
   struct channel_config pri_cf = {
     .name = "pri",
     .channel = cc->channel,
     .table = cc->table,
     .out_filter = cc->out_filter,
     .in_limit = cc->in_limit,
-    .ra_mode = cc->ra_mode,
+    .ra_mode = RA_ANY,
-    .ai_aggr = cc->ai_aggr,
     .debug = cc->debug,
     .rpki_reload = cc->rpki_reload,
   };
@@ -163,30 +160,6 @@ pipe_configure_channels(struct pipe_proto *p, struct pipe_config *cf)
     .rpki_reload = cc->rpki_reload,
   };
 
-  log("ai_aggr = %p", cc->ai_aggr);
-  const struct aggr_item_linearized *ail = cc->ai_aggr;
-  int node = 1;
-
-  if (ail != NULL) {
-    for (int i = 0; i < ail->count; i++) {
-      switch (ail->items[i].type) {
-        case AGGR_ITEM_TERM:
-          log("node %d, type: term", node);
-          break;
-        case AGGR_ITEM_STATIC_ATTR:
-          log("node %d, type: static", node);
-          break;
-        case AGGR_ITEM_DYNAMIC_ATTR:
-          log("node %d, type: dynamic", node);
-          break;
-        default:
-          log("node %d, type: other", node);
-          break;
-      }
-      node++;
-    }
-  }
-
   return
     proto_configure_channel(&p->p, &p->pri, &pri_cf) &&
     proto_configure_channel(&p->p, &p->sec, &sec_cf);

proto/pipe/pipe.h

@@ -9,21 +9,9 @@
 #ifndef _BIRD_PIPE_H_
 #define _BIRD_PIPE_H_
 
-struct merging_import {
-  uint limit;
-  uint use_aggregator;
-};
-
-struct merging_export {
-  uint limit;
-  uint use_aggregator;
-};
-
 struct pipe_config {
   struct proto_config c;
-  struct rtable_config *peer;		    /* Table we're connected to */
+  struct rtable_config *peer;		/* Table we're connected to */
-  struct merging_import config_import;  /* From peer table to primary table */
-  struct merging_export config_export;  /* From primary table to peer table */
 };
 
 struct pipe_proto {

sysdep/unix/krt.c

@@ -581,9 +581,6 @@ krt_export_net(struct krt_proto *p, net *net, rte **rt_free)
   if (c->ra_mode == RA_MERGED)
     return rt_export_merged(c, net, rt_free, krt_filter_lp, 1);
 
-  if (c->ra_mode == RA_AGGREGATED)
-    return rt_export_aggregated(c, net, rt_free, krt_filter_lp, 1);
-
   rt = net->routes;
   *rt_free = NULL;