bird/proto/bmp/bmp.c

/*
 *	BIRD -- The BGP Monitoring Protocol (BMP)
 *
 *	(c) 2020 Akamai Technologies, Inc. (Pawel Maslanka, pmaslank@akamai.com)
 *
 *	Can be freely distributed and used under the terms of the GNU GPL.
 */

/**
 * DOC: BGP Monitoring Protocol (BMP)
 *
 * Supported standards:
 * o RFC 7854 - BMP standard
 *
 * TODO:
 * - Support Peer Distinguisher ID in Per-Peer Header
 * - Support peer type as RD Instance in Peer Type field of Per-Peer Header.
 *   Currently, there are supported Global and Local Instance Peer types
 * - Support corresponding FSM event code during send PEER DOWN NOTIFICATION
 * - Support DE_CONFIGURED PEER DOWN REASON code in PEER DOWN NOTIFICATION message
 * - If connection with BMP collector will lost then we don't establish connection again
 * - Set Peer Type by its a global and local-scope IP address
 *
 * The BMP session is managed by a simple state machine with three states: Idle
 * (!started, !sk), Connect (!started, sk active), and Established (started). It
 * has three events: connect successful (Connect -> Established), socket error
 * (any -> Idle), and connect timeout (Idle/Connect -> Connect, resetting the
 * TCP socket).
 */

#include "proto/bmp/bmp.h"
#include "proto/bmp/buffer.h"
#include "proto/bmp/map.h"

#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <netdb.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <arpa/inet.h>
#include <limits.h>

#include "nest/cli.h"
#include "filter/filter.h"
#include "proto/bgp/bgp.h"
#include "sysdep/unix/unix.h"
#include "sysdep/unix/io-loop.h"
#include "lib/event.h"
#include "lib/ip.h"
#include "lib/lists.h"
#include "lib/resource.h"
#include "lib/unaligned.h"
#include "lib/tlists.h"
#include "nest/iface.h"
#include "nest/route.h"

#define HASH_PEER_KEY(n)		n->bgp
#define HASH_PEER_NEXT(n)		n->next
#define HASH_PEER_EQ(b1,b2)		b1 == b2
#define HASH_PEER_FN(b)			ptr_hash(b)

#define BMP_STREAM_KEY_POLICY		0x100

#define HASH_STREAM_KEY(n)		n->bgp, n->key
#define HASH_STREAM_NEXT(n)		n->next
#define HASH_STREAM_EQ(b1,k1,b2,k2)	b1 == b2 && k1 == k2
#define HASH_STREAM_FN(b,k)		ptr_hash(b) ^ u32_hash(k)

#define HASH_TABLE_KEY(n)		n->table
#define HASH_TABLE_NEXT(n)		n->next
#define HASH_TABLE_EQ(t1,t2)		t1 == t2
#define HASH_TABLE_FN(t)		ptr_hash(t)

/* BMP Common Header [RFC 7854 - Section 4.1] */
enum bmp_version {
  BMP_VER_UNUSED = 0, // Version 0 is reserved and MUST NOT be sent
  BMP_VERSION_1 = 1,  // Version 1 was used by draft version of RFC 7854
  BMP_VERSION_2 = 2,  // Version 2 was used by draft version of RFC 7854
  BMP_VERSION_3 = 3   // Version 3 is used by all messages defined in RFC 7854
};

enum bmp_message_type {
  BMP_ROUTE_MONITOR = 0,   // Route Monitoring
  BMP_STATS_REPORT = 1,    // Statistics Report
  BMP_PEER_DOWN_NOTIF = 2, // Peer Down Notification
  BMP_PEER_UP_NOTIF = 3,   // Peer Up Notification
  BMP_INIT_MSG = 4,        // Initiation Message
  BMP_TERM_MSG = 5,        // Termination Message
  BMP_ROUTE_MIRROR_MSG = 6 // Route Mirroring Message
};

// Total size of Common Header
#define BMP_COMMON_HDR_SIZE 6
// Defines size of padding when IPv4 address is going to be put into field
// which can accept also IPv6 address
#define BMP_PADDING_IP4_ADDR_SIZE 12

/* BMP Per-Peer Header [RFC 7854 - Section 4.2] */
// Total size of Per-Peer Header
#define BMP_PER_PEER_HDR_SIZE 42

enum bmp_peer_type {
  BMP_PEER_TYPE_GLOBAL_INSTANCE = 0,
  BMP_PEER_TYPE_RD_INSTANCE = 1,
  BMP_PEER_TYPE_LOCAL_INSTANCE = 2
};

#define BMP_PEER_HDR_FLAG_V_SHIFT 7
enum bmp_peer_flag_v_t {
  // The Peer address is an IPv4 address
  BMP_PEER_HDR_FLAG_V_IP4 = (0 << BMP_PEER_HDR_FLAG_V_SHIFT),
  // The Peer address is an IPv6 address
  BMP_PEER_HDR_FLAG_V_IP6 = (1 << BMP_PEER_HDR_FLAG_V_SHIFT)
};

#define BMP_PEER_HDR_FLAG_L_SHIFT 6
enum bmp_peer_flag_l {
  BMP_PEER_HDR_FLAG_L_PRE_POLICY_ADJ_RIB_IN = (0 << BMP_PEER_HDR_FLAG_L_SHIFT),
  BMP_PEER_HDR_FLAG_L_POST_POLICY_ADJ_RIB_IN = (1 << BMP_PEER_HDR_FLAG_L_SHIFT)
};

#define BMP_PEER_HDR_FLAG_A_SHIFT 5
enum bmp_peer_flag_a {
  // The 4-byte AS_PATH format
  BMP_PEER_HDR_FLAG_A_AS_PATH_4B = (0 << BMP_PEER_HDR_FLAG_A_SHIFT),
  // The legacy 2-byte AS_PATH format
  BMP_PEER_HDR_FLAG_A_AS_PATH_2B = (1 << BMP_PEER_HDR_FLAG_A_SHIFT)
};

#define BMP_PEER_HDR_FLAGS_INIT(flags) \
  (flags) = 0
#define BMP_PEER_HDR_FLAGS_SET(flags, bit_mask) \
  (flags) |= (bit_mask)

/* BMP Information TLV header [RFC 7854 - Section 4.4] */
// Total size of Type and Length fields of Information TLV Header without
// variable part
#define BMP_INFO_TLV_FIX_SIZE 4

enum bmp_info_tlv_type {
  BMP_INFO_TLV_TYPE_STRING = 0,    // String
  BMP_INFO_TLV_TYPE_SYS_DESCR = 1, // SysDescr
  BMP_INFO_TLV_TYPE_SYS_NAME = 2   // SysName
};

/* BMP Peer Up Notification message header [RFC 7854 - Section 4.10] */
// Total size of all fields of Peer Up Notification message except variable part
#define BMP_PEER_UP_NOTIF_MSG_FIX_SIZE 20

enum bmp_peer_down_notif_reason {
  // The local system closed the session
  BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION = 1,
  // The local system closed the session
  BMP_PEER_DOWN_REASON_LOCAL_NO_NOTIFICATION = 2,
  // The remote system closed the session with a notification message
  BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION = 3,
  // The remote system closed the session without a notification message
  BMP_PEER_DOWN_REASON_REMOTE_NO_NOTIFICATION = 4,
  // Information for this peer will no longer be sent to the monitoring station
  // for configuration reasons
  BMP_PEER_DOWN_REASON_PEER_DE_CONFIGURED = 5
};

/* BMP Termination Message [RFC 7854 - Section 4.5] */
#define BMP_TERM_INFO_TYPE_SIZE 2
enum bmp_term_info_type {
  BMP_TERM_INFO_STRING = 0, // The Information field contains string
  BMP_TERM_INFO_REASON = 1, // The Information field contains 2-byte reason code
};

// 2-byte code in the Information field
#define BMP_TERM_REASON_CODE_SIZE 2
enum bmp_term_reason {
  BMP_TERM_REASON_ADM = 0,  // Session administratively closed
  BMP_TERM_REASON_UNK = 1,  // Unspecified reason
  BMP_TERM_REASON_OOR = 2,  // Out of resources
  BMP_TERM_REASON_DUP = 3,  // Redundant connection
  BMP_TERM_REASON_PERM = 4, // Session permanently administratively closed
};

// Size of Information Length field in Termination Message header
#define BMP_TERM_INFO_LEN_FIELD_SIZE 2

// Default chunk size request when memory allocation
#define DEFAULT_MEM_BLOCK_SIZE 4096

// Initial delay for connection to the BMP collector
#define CONNECT_INIT_TIME (200 MS)

// Timeout for connection to the BMP collector retry
#define CONNECT_RETRY_TIME (10 S)

#define IP4_MAX_TTL 255


#define IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(expr, msg, rv...)     \
  do {                                                                      \
    if ((expr))                                                             \
    {                                                                       \
      log(L_WARN "[BMP] " msg);                                             \
      return rv;                                                            \
    }                                                                       \
  } while (0)


#define IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(p, msg, rv...)	\
  do {									\
    IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(!(p), msg, rv);	\
  } while (0)

static const struct ea_class *bgp_next_hop_ea_class = NULL;

static void bmp_connected(struct birdsock *sk);
static void bmp_sock_err(sock *sk, int err);
static void bmp_close_socket(struct bmp_proto *p);
static void bmp_check_routes(void *bt_);
static void bmp_feed_end(struct rt_export_request *req);

static void
bmp_send_peer_up_notif_msg(struct bmp_proto *p, ea_list *bgp,
  const byte *tx_data, const size_t tx_data_size,
  const byte *rx_data, const size_t rx_data_size);

static void bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs);

// Stores necessary any data in list
struct bmp_data_node {
  node n;
  byte *data;
  size_t data_size;

  u32 remote_as;
  u32 remote_id;
  ip_addr remote_ip;
  btime timestamp;
  bool global_peer;
  bool policy;
};

static void
bmp_common_hdr_serialize(buffer *stream, const enum bmp_message_type type, const u32 data_size)
{
  bmp_put_u8(stream, BMP_VERSION_3);
  bmp_put_u32(stream, BMP_COMMON_HDR_SIZE + data_size);
  bmp_put_u8(stream, type);
}

static void
bmp_info_tlv_hdr_serialize(buffer *stream, const enum bmp_info_tlv_type type,
  const char *str)
{
  size_t str_len = strlen(str);
  str_len = MIN(str_len, MIB_II_STR_LEN);

  bmp_put_u16(stream, type);
  bmp_put_u16(stream, str_len);
  bmp_put_data(stream, str, str_len);
}

// Serializes BMP Initiation message header [RFC 7854 - Section 4.3]
static void
bmp_init_msg_serialize(buffer *stream, const char *sys_descr, const char *sys_name)
{
  const size_t sys_descr_len = strlen(sys_descr);
  const size_t sys_name_len = strlen(sys_name);
  // We include MIB-II sysDescr and sysName in BMP INIT MSG so that's why
  // allocated 2x BMP_INFO_TLV_FIX_SIZE memory pool size
  const size_t data_size = (2 * BMP_INFO_TLV_FIX_SIZE) + sys_descr_len + sys_name_len;
  bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
  bmp_common_hdr_serialize(stream, BMP_INIT_MSG, data_size);
  bmp_info_tlv_hdr_serialize(stream, BMP_INFO_TLV_TYPE_SYS_DESCR, sys_descr);
  bmp_info_tlv_hdr_serialize(stream, BMP_INFO_TLV_TYPE_SYS_NAME, sys_name);
}

static void
bmp_schedule_tx_packet(struct bmp_proto *p, const byte *payload, const size_t size)
{
  ASSERT(p->started);

  struct bmp_data_node *tx_data = mb_allocz(p->tx_mem_pool, sizeof (struct bmp_data_node));
  tx_data->data = mb_allocz(p->tx_mem_pool, size);
  log("tx_data %x data %x", tx_data, tx_data->data);
  memcpy(tx_data->data, payload, size);
  tx_data->data_size = size;
  log("schedule tx packet size %i", size);
  if (size > 5000)
      log("too big");
  add_tail(&p->tx_queue, &tx_data->n);

  if (sk_tx_buffer_empty(p->sk)
      && !ev_active(p->tx_ev))
  {
    ev_schedule(p->tx_ev);
  }
}

static void
bmp_fire_tx(void *p_)
{
  log("fire");
  struct bmp_proto *p = p_;

  if (!p->started)
    return;

  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
    EMPTY_LIST(p->tx_queue),
    "Called BMP TX event handler when there is not any data to send"
  );

  size_t cnt = 0; // Counts max packets which we want to send per TX slot
  struct bmp_data_node *tx_data;
  struct bmp_data_node *tx_data_next;
  WALK_LIST_DELSAFE(tx_data, tx_data_next, p->tx_queue)
  {
    log("WALK_LIST_DELSAFE");
    if (tx_data->data_size > p->sk->tbsize)
    {
      sk_set_tbsize(p->sk, tx_data->data_size);
    }

    size_t data_size = tx_data->data_size;
    memcpy(p->sk->tbuf, tx_data->data, data_size);
    mb_free(tx_data->data);
    rem_node((node *) tx_data);
    mb_free(tx_data);
    
    log("walk ifs for returns socket %i, data size %i", p->sk, data_size);

    if (sk_send(p->sk, data_size) <= 0)
      return;

    // BMP packets should be treat with lowest priority when scheduling sending
    // packets to target. That's why we want to send max. 32 packets per event
    // call
    if (++cnt > 32)
    {
      if (!ev_active(p->tx_ev))
      {
        ev_schedule(p->tx_ev);
      }

      return;
    }
  }
}

static void
bmp_tx(struct birdsock *sk)
{
  bmp_fire_tx(sk->data);
}

/* We need RX hook just to accept socket close events */
static int
bmp_rx(struct birdsock *sk UNUSED, uint size UNUSED)
{
  return 0;
}


static inline void
bmp_put_ipa(buffer *stream, const ip_addr addr)
{
  bmp_put_ip6(stream, ipa_is_ip4(addr) ?
	      ip6_build(0,0,0, ipa_to_u32(addr)) :
	      ipa_to_ip6(addr));
}

static void
bmp_put_bgp_hdr(buffer *stream, const u8 msg_type, const u16 msg_length)
{
  bmp_buffer_need(stream, BGP_HEADER_LENGTH);

  memset(stream->pos, 0xff, BGP_HDR_MARKER_LENGTH);
  stream->pos += BGP_HDR_MARKER_LENGTH;

  bmp_put_u16(stream, msg_length);
  bmp_put_u8(stream, msg_type);
}

/**
 * bmp_per_peer_hdr_serialize - serializes Per-Peer Header
 *
 * @is_post_policy: indicate the message reflects the post-policy Adj-RIB-In
 * @peer_addr: the remote IP address associated with the TCP session
 * @peer_as: the Autonomous System number of the peer
 * @peer_bgp_id: the BGP Identifier of the peer
 * @ts_sec: the time in seconds when the encapsulated routes were received
 * @ts_usec: the time in microseconds when the encapsulated routes were received
 */
static void
bmp_per_peer_hdr_serialize(buffer *stream, const bool is_global_instance_peer,
  const bool is_post_policy, const bool is_as_path_4bytes,
  const ip_addr peer_addr, const u32 peer_as, const u32 peer_bgp_id,
  const u32 ts_sec, const u32 ts_usec)
{
  // TODO: ATM we don't support BMP_PEER_TYPE_RD_INSTANCE
  const enum bmp_peer_type peer_type = is_global_instance_peer
                                      ? BMP_PEER_TYPE_GLOBAL_INSTANCE
                                      : BMP_PEER_TYPE_LOCAL_INSTANCE;
  const u8 peer_flag_v = ipa_is_ip4(peer_addr)
                           ? BMP_PEER_HDR_FLAG_V_IP4
                           : BMP_PEER_HDR_FLAG_V_IP6;
  const u8 peer_flag_l = is_post_policy
                           ? BMP_PEER_HDR_FLAG_L_POST_POLICY_ADJ_RIB_IN
                           : BMP_PEER_HDR_FLAG_L_PRE_POLICY_ADJ_RIB_IN;
  const u8 peer_flag_a = is_as_path_4bytes
                           ? BMP_PEER_HDR_FLAG_A_AS_PATH_4B
                           : BMP_PEER_HDR_FLAG_A_AS_PATH_2B;
  u8 peer_flags;
  BMP_PEER_HDR_FLAGS_INIT(peer_flags);
  BMP_PEER_HDR_FLAGS_SET(peer_flags, peer_flag_v);
  BMP_PEER_HDR_FLAGS_SET(peer_flags, peer_flag_l);
  BMP_PEER_HDR_FLAGS_SET(peer_flags, peer_flag_a);

  bmp_put_u8(stream, peer_type);
  bmp_put_u8(stream, peer_flags);
  // TODO: Provide appropriate peer Route Distinguisher if applicable
  bmp_put_u64(stream, 0x00); // 0x00 - Not supported peer distinguisher
  bmp_put_ipa(stream, peer_addr);
  bmp_put_u32(stream, peer_as);
  bmp_put_u32(stream, peer_bgp_id);
  bmp_put_u32(stream, ts_sec);
  bmp_put_u32(stream, ts_usec);
}

/* [4.6] Route Monitoring */
static void
bmp_route_monitor_msg_serialize(buffer *stream, const bool is_peer_global,
  const bool table_in_post_policy, const u32 peer_as, const u32 peer_bgp_id,
  const bool as4_support, const ip_addr remote_addr, const byte *update_msg,
  const size_t update_msg_size, btime timestamp)
{
  const size_t data_size = BMP_PER_PEER_HDR_SIZE + update_msg_size;
  u32 ts_sec = timestamp TO_S;
  u32 ts_usec = timestamp - (ts_sec S);

  bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
  bmp_common_hdr_serialize(stream, BMP_ROUTE_MONITOR, data_size);
  bmp_per_peer_hdr_serialize(stream, is_peer_global, table_in_post_policy,
    as4_support, remote_addr, peer_as, peer_bgp_id, ts_sec, ts_usec);
  bmp_put_data(stream, update_msg, update_msg_size);
}

static void
bmp_peer_up_notif_msg_serialize(buffer *stream, const bool is_peer_global,
  const u32 peer_as, const u32 peer_bgp_id, const bool as4_support,
  const ip_addr local_addr, const ip_addr remote_addr, const u16 local_port,
  const u16 remote_port, const byte *sent_msg, const size_t sent_msg_length,
  const byte *recv_msg, const size_t recv_msg_length)
{
  const size_t data_size =
    BMP_PER_PEER_HDR_SIZE + BMP_PEER_UP_NOTIF_MSG_FIX_SIZE +
    BGP_HEADER_LENGTH + sent_msg_length + BGP_HEADER_LENGTH + recv_msg_length;

  bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
  bmp_common_hdr_serialize(stream, BMP_PEER_UP_NOTIF, data_size);
  bmp_per_peer_hdr_serialize(stream, is_peer_global,
    false /* TODO: Hardcoded pre-policy Adj-RIB-In */, as4_support, remote_addr,
    peer_as, peer_bgp_id, 0, 0); // 0, 0 - No timestamp provided
  bmp_put_ipa(stream, local_addr);
  bmp_put_u16(stream, local_port);
  bmp_put_u16(stream, remote_port);
  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + sent_msg_length);
  bmp_put_data(stream, sent_msg, sent_msg_length);
  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + recv_msg_length);
  bmp_put_data(stream, recv_msg, recv_msg_length);
}

static void
bmp_peer_down_notif_msg_serialize(buffer *stream, const bool is_peer_global,
  const u32 peer_as, const u32 peer_bgp_id, const bool as4_support,
  const ip_addr remote_addr, const byte *data, const size_t data_size)
{
  const size_t payload_size = BMP_PER_PEER_HDR_SIZE + data_size;
  bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + payload_size);
  bmp_common_hdr_serialize(stream, BMP_PEER_DOWN_NOTIF, payload_size);
  bmp_per_peer_hdr_serialize(stream, is_peer_global,
    false /* TODO: Hardcoded pre-policy adj RIB IN */,  as4_support, remote_addr,
    peer_as, peer_bgp_id, 0, 0); // 0, 0 - No timestamp provided
  bmp_put_data(stream, data, data_size);
}


/*
 *	BMP tables
 */

static struct bmp_table *
bmp_find_table(struct bmp_proto *p, rtable *tab)
{
  return HASH_FIND(p->table_map, HASH_TABLE, tab);
}

const struct channel_class channel_bmp = {
  .channel_size =	sizeof(struct channel),
  .config_size =	sizeof(struct channel_config),
  /*.init =		
  .start =		
  .shutdown =		
  .cleanup =		
  .reconfigure =	*/
};

static struct bmp_table *
bmp_add_table(struct bmp_proto *p, rtable *tab)
{
  struct bmp_table *bt = mb_allocz(p->p.pool, sizeof(struct bmp_table));
  log("adding table %x", bt);
  bt->table = tab;
  bt->p = p;
  rt_lock_table(bt->table);

  HASH_INSERT(p->table_map, HASH_TABLE, bt);

  bt->event.data = bt;
  log("name %s post %i pre %i", p->p.name, p->monitoring_rib.in_post_policy, p->monitoring_rib.in_pre_policy);

  //if(p->monitoring_rib.in_post_policy){
  bt->event.hook = bmp_check_routes;
  bt->out_req = (struct rt_export_request) {
    .name = mb_sprintf(p->p.pool, "%s.exp_request", p->p.name),
    .r = (struct lfjour_recipient) {
      .target = proto_event_list(&p->p),//&p->p.loop->event_list,
      .event = &bt->event,
    },
    .pool = p->p.pool,
    .trace_routes = p->p.debug,
    //.dump = channel_dump_export_req, TODO: this will crash on `dump tables` from CLI
    .fed = bmp_feed_end,
  };
  rt_export_subscribe(tab, all, &bt->out_req);//}
  return bt;
}

/*static void
bmp_remove_table(struct bmp_proto *p, struct bmp_table *bt)
{
  log("removing table - bmp table %x chann %x, (subscr %x uc %x)",bt,  bt->channel, &bt->channel->roa_subscriptions, &bt->uc);
  channel_set_state(bt->channel, CS_STOP);
  channel_set_state(bt->channel, CS_DOWN);
  proto_remove_channel(&p->p, bt->channel);

  HASH_REMOVE(p->table_map, HASH_TABLE, bt);

  rt_unlock_table(bt->table);
  bt->table = NULL;

  log("free table %x", bt);
  mb_free(bt);
}*/

static void
bmp_remove_table_rt(struct bmp_proto *p, struct bmp_table *bt) // still falling in krt routes... The thread synchronization was maybe not needed...
{
  log("removing table - bmp table %x chann %x, (subscr %x uc %x)",bt,  bt->channel, &bt->channel->roa_subscriptions, &bt->uc);
  if (bt->channel)
  {
    channel_set_state(bt->channel, CS_STOP);
    channel_set_state(bt->channel, CS_DOWN);
  }
  //if (p->monitoring_rib.in_post_policy)
    rt_export_unsubscribe(all, &bt->out_req);
  //else
    //rt_feeder_unsubscribe(&bt->in_req);

  HASH_REMOVE(p->table_map, HASH_TABLE, bt);

  rt_unlock_table(bt->table);
  bt->table = NULL;

  log("free table %x", bt);
  mb_free(bt);
  //log("free done");
}

static inline void bmp_lock_table(struct bmp_proto *p UNUSED, struct bmp_table *bt)
{ bt->uc++; }

struct bmp_table *bmp_get_table(struct bmp_proto *p, rtable *tab)
{
  struct bmp_table *bt = bmp_find_table(p, tab);
  if (bt)
  {
   while (true) {
      atomic_int i = bt->uc;
      if (i == 0)
      {
        struct bmp_table *new = bmp_add_table(p, tab);
        bmp_lock_table(p, new);
        return new;
      }
      if (atomic_compare_exchange_strong_explicit(&bt->uc, &i, i+1, memory_order_acq_rel, memory_order_relaxed))
        return bt;
    }
  }
  struct bmp_table *new = bmp_add_table(p, tab);
  bmp_lock_table(p, new);
  return new;
}

static inline void bmp_unlock_table(struct bmp_proto *p, struct bmp_table *bt)
{ atomic_int i = 1;
  if (atomic_compare_exchange_strong_explicit(&bt->uc, &i, 0, memory_order_acq_rel, memory_order_relaxed))
    bmp_remove_table_rt(p, bt);
  else
    bt->uc--;
}


/*
 *	BMP streams
 */

static inline u32 bmp_stream_key(u32 afi, bool policy)
{ return afi ^ (policy ? BMP_STREAM_KEY_POLICY : 0); }

//static inline u32 bmp_stream_afi(struct bmp_stream *bs)
//{ return bs->key & ~BMP_STREAM_KEY_POLICY; }

static inline bool bmp_stream_policy(struct bmp_stream *bs)
{ return !!(bs->key & BMP_STREAM_KEY_POLICY); }

static struct bmp_stream *
bmp_find_stream(struct bmp_proto *p, const struct bgp_proto *bgp, u32 afi, bool policy)
{
  ea_list *bgp_attr = proto_state_table->attrs[bgp->p.id];
  return HASH_FIND(p->stream_map, HASH_STREAM, bgp_attr, bmp_stream_key(afi, policy));

}

static struct bmp_stream *
bmp_add_stream(struct bmp_proto *p, struct bmp_peer *bp, u32 afi, bool policy, rtable *tab, ea_list *sender, int in_pre_policy)
{
  struct bmp_stream *bs = mb_allocz(p->p.pool, sizeof(struct bmp_stream));
  log("add stream to p %i bs %x", p, bs);
  bs->bgp = bp->bgp;
  bs->key = bmp_stream_key(afi, policy);

  add_tail(&bp->streams, &bs->n);
  HASH_INSERT(p->stream_map, HASH_STREAM, bs);

  bs->table = bmp_get_table(p, tab);
  //bmp_lock_table(p, bs->table);

  bs->sender = sender;
  bs->sync = false;
  bs->in_pre_policy = in_pre_policy;

  return bs;
}

static void
bmp_remove_stream(struct bmp_proto *p, struct bmp_stream *bs)
{
  log("go to unlock table");
  rem_node(&bs->n);
  HASH_REMOVE(p->stream_map, HASH_STREAM, bs);

  bmp_unlock_table(p, bs->table);
  bs->table = NULL;

  mb_free(bs);
}


/*
 *	BMP peers
 */

static struct bmp_peer *
bmp_find_peer(struct bmp_proto *p, ea_list *bgp_attr)
{
  return HASH_FIND(p->peer_map, HASH_PEER, bgp_attr);
}

static struct bmp_peer *
bmp_add_peer(struct bmp_proto *p, ea_list *bgp_attr)
{
  log("domain is locked %i, service is locked %i", DG_IS_LOCKED(p->p.pool->domain), locking_stack.service);
  struct bmp_peer *bp;
  if (DG_IS_LOCKED(p->p.pool->domain))
    bp = mb_allocz(p->p.pool, sizeof(struct bmp_peer));
  else
  {
    DG_LOCK(p->p.pool->domain);
    bp = mb_allocz(p->p.pool, sizeof(struct bmp_peer));
    DG_UNLOCK(p->p.pool->domain);
  }
  log("bmp_peer %x", bp);
  bp->bgp = bgp_attr;

  init_list(&bp->streams);

  HASH_INSERT(p->peer_map, HASH_PEER, bp);

  int proto_id = ea_get_int(bgp_attr, &ea_proto_id, 0);
 
 
  struct channel_attrs *chan_attr;
  log("before while id %i, eattrs %i", proto_id, proto_state_table->channels_attrs[proto_id]);
  WALK_TLIST(channel_attrs, chan_attr, &proto_state_table->channels_attrs[proto_id])
  {
    log("chan_attr in bmp_add_peer %i, attrs %i", chan_attr, chan_attr->attrs);
    rtable *ch_table = (rtable *) ea_get_ptr(chan_attr->attrs, &ea_rtable, 0);
    const char *name = ea_get_adata(chan_attr->attrs, &ea_name)->data;
    int in_keep = ea_get_int(chan_attr->attrs, &ea_in_keep, 0);

    log("name %s  ch_table %i first if %i second if %i in keep %i", name, ch_table, p->monitoring_rib.in_pre_policy, p->monitoring_rib.in_post_policy, in_keep);


    if (p->monitoring_rib.in_pre_policy && ch_table)
    {
      log("in pre");
      if (in_keep == RIK_PREFILTER)
      {
        log("add stream in keep %i", in_keep);
        bmp_add_stream(p, bp, ea_get_int(chan_attr->attrs, &ea_bgp_afi, 0), false, ch_table, chan_attr->attrs, 1);
        }
      else
        log(L_WARN "%s: Try to do pre policy with disabled import tables (channel %s)", p->p.name, name);
    }

    if (p->monitoring_rib.in_post_policy && ch_table)
    {
      log("in post");
      bmp_add_stream(p, bp, ea_get_int(chan_attr->attrs, &ea_bgp_afi, 0), true, ch_table, chan_attr->attrs, 0);
    }
  }

  return bp;
}

static void
bmp_remove_peer(struct bmp_proto *p, struct bmp_peer *bp)
{
  struct bmp_stream *bs, *bs_next;
  WALK_LIST_DELSAFE(bs, bs_next, bp->streams)
    bmp_remove_stream(p, bs);//TODO//TODO//TODO

  HASH_REMOVE(p->peer_map, HASH_PEER, bp);

  mb_free(bp);
}

static void
bmp_peer_up_(struct bmp_proto *p, ea_list *bgp_attr, bool sync,
	    const byte *tx_open_msg, uint tx_open_length,
	    const byte *rx_open_msg, uint rx_open_length)
{
  if (!p->started)
    return;

  struct bmp_peer *bp = bmp_find_peer(p, bgp_attr);
  if (bp)
    return;

  const char *name = ea_get_adata(bgp_attr, &ea_name)->data;
  TRACE(D_STATES, "Peer up for %s", name);

  bp = bmp_add_peer(p, bgp_attr);

  bmp_send_peer_up_notif_msg(p, bgp_attr, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);

  /*
   * We asssume peer_up() notifications are received before any route
   * notifications from that peer. Therefore, peers established after BMP
   * session coould be considered synced with empty RIB.
   */
  if (sync)
  {
    struct bmp_stream *bs;
    WALK_LIST(bs, bp->streams)
    {
      bmp_route_monitor_end_of_rib(p, bs);
      bs->sync = true;
    }
  }
}

//void
//bmp_peer_up(ea_list *bgp,
//	    const byte *tx_open_msg, uint tx_open_length,
//	    const byte *rx_open_msg, uint rx_open_length)
//{
//  struct bmp_proto *p; node *n;
//  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
//    bmp_peer_up_(p, bgp, true, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
//}

static void
bmp_peer_init(struct bmp_proto *p, ea_list *bgp_attr)
{
  const struct bgp_conn *conn = (const struct bgp_conn *) ea_get_ptr(bgp_attr, &ea_bgp_conn, 0);

  if (!conn || (conn->state != BS_ESTABLISHED) ||
      !conn->local_open_msg || !conn->remote_open_msg)
    return;

  bmp_peer_up_(p, bgp_attr, false, conn->local_open_msg, conn->local_open_length,
	       conn->remote_open_msg, conn->remote_open_length);
}



static const struct birdsock *
bmp_get_birdsock(ea_list *bgp)
{
  struct bgp_conn *conn = (struct bgp_conn *) ea_get_ptr(bgp, &ea_bgp_conn, 0);
  if (conn && conn->sk)
    return conn->sk;

  return NULL;
}

static const struct birdsock *
bmp_get_birdsock_ext(ea_list *bgp)
{
  const struct birdsock *sk = bmp_get_birdsock(bgp);
  if (sk != NULL)
    return sk;

  struct bgp_conn *in_conn = (struct bgp_conn *) ea_get_ptr(bgp, &ea_bgp_in_conn, 0);
  struct bgp_conn *out_conn = (struct bgp_conn *) ea_get_ptr(bgp, &ea_bgp_out_conn, 0);
  if (in_conn->sk)
  {
    sk = in_conn->sk;
  }
  else if (out_conn->sk)
  {
    sk = out_conn->sk;
  }

  return sk;
}

static const struct bgp_caps *
bmp_get_bgp_remote_caps(struct bgp_conn *bgp_conn)
{
  if (bgp_conn && bgp_conn->remote_caps)
    return bgp_conn->remote_caps;

  return NULL;
}

static const struct bgp_caps *
bmp_get_bgp_remote_caps_ext(ea_list *bgp)
{
  struct bgp_conn *bgp_conn = (struct bgp_conn*)  ea_get_adata(bgp, &ea_bgp_conn)->data;
  const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps(bgp_conn);
  if (remote_caps != NULL)
    return remote_caps;

  struct bgp_conn *bgp_in_conn = (struct bgp_conn*)  ea_get_adata(bgp, &ea_bgp_in_conn)->data;
  struct bgp_conn *bgp_out_conn = (struct bgp_conn*)  ea_get_adata(bgp, &ea_bgp_out_conn)->data;
  if (bgp_in_conn->remote_caps)
  {
    remote_caps = bgp_in_conn->remote_caps;
  }
  else if (bgp_out_conn->remote_caps)
  {
    remote_caps = bgp_out_conn->remote_caps;
  }

  return remote_caps;
}

static bool
bmp_is_peer_global_instance(ea_list *bgp)
{
  int peer_type = ea_get_int(bgp, &ea_bgp_peer_type, 0);
  int local_as = ea_get_int(bgp, &ea_bgp_loc_as, 0);
  int remote_as = ea_get_int(bgp, &ea_bgp_rem_as, 0);
  log("bmp_is_peer_global_instance loc as %i rem as %i peer_type %i (constants BGP_PT_EXTERNAL %i, BGP_PT_INTERNAL %i)", local_as, remote_as, peer_type, BGP_PT_EXTERNAL, BGP_PT_INTERNAL);
  return (peer_type != BGP_PT_EXTERNAL &&
            peer_type != BGP_PT_INTERNAL)
              ? (local_as != remote_as)
              : (peer_type == BGP_PT_EXTERNAL);
}

static void
bmp_send_peer_up_notif_msg(struct bmp_proto *p, ea_list *bgp,
  const byte *tx_data, const size_t tx_data_size,
  const byte *rx_data, const size_t rx_data_size)
{
  ASSERT(p->started);

  const struct birdsock *sk = bmp_get_birdsock_ext(bgp);
  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
    sk,
    "[BMP] No BGP socket"
  );

  const int rem_as = ea_get_int(bgp, &ea_bgp_rem_as, 0);
  const int rem_id = ea_get_int(bgp, &ea_bgp_rem_id, 0);
  log("bmp_send_peer_up_notif_msg rem_as %i rem_id %i", rem_as, rem_id);
  const bool is_global_instance_peer = bmp_is_peer_global_instance(bgp);
  buffer payload = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
  bmp_peer_up_notif_msg_serialize(&payload, is_global_instance_peer,
    rem_as, rem_id, 1,
    sk->saddr, sk->daddr, sk->sport, sk->dport, tx_data, tx_data_size,
    rx_data, rx_data_size);
  bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
  bmp_buffer_free(&payload);
}

static void
bmp_route_monitor_put_update(struct bmp_proto *p, struct bmp_stream *bs, const byte *data, size_t length, btime timestamp)
{
  struct bmp_data_node *upd_msg = mb_allocz(p->update_msg_mem_pool,
                               sizeof (struct bmp_data_node));
  upd_msg->data = mb_alloc(p->update_msg_mem_pool, length);
  log("upd_msg %x, data %x", upd_msg, upd_msg->data);
  memcpy(upd_msg->data, data, length);
  upd_msg->data_size = length;

  add_tail(&p->update_msg_queue, &upd_msg->n);

  /* Save some metadata */
  ea_list *bgp = bs->bgp;
  upd_msg->remote_as = ea_get_int(bgp, &ea_bgp_rem_as, 0);
  upd_msg->remote_id = ea_get_int(bgp, &ea_bgp_rem_id, 0);
  upd_msg->remote_ip = ea_get_ip(bgp, &ea_bgp_rem_ip, IPA_NONE);
  upd_msg->timestamp = timestamp;
  upd_msg->global_peer = bmp_is_peer_global_instance(bgp);
  upd_msg->policy = bmp_stream_policy(bs);

  /* Kick the commit */
  if (!ev_active(p->update_ev))
    ev_schedule(p->update_ev);
}

static void
bmp_route_monitor_notify(struct bmp_proto *p, struct bgp_proto *bgp_p, struct bmp_stream *bs, const struct rte *new)
{
  log("notified");
      bmp_route_monitor_end_of_rib(p, bs);
  byte buf[BGP_MAX_EXT_MSG_LENGTH];
  byte *end = bgp_bmp_encode_rte(bs->sender, bgp_p, buf, new);

  btime delta_t = new->attrs ? current_time() - new->lastmod : 0;
  btime timestamp = current_real_time() - delta_t;

  if (end)
    bmp_route_monitor_put_update(p, bs, buf, end - buf, timestamp);
  else
    log(L_WARN "%s: Cannot encode update for %N", p->p.name, new->net);
}

static void
bmp_route_monitor_commit(void *p_)
{
  struct bmp_proto *p = p_;

  if (!p->started)
    return;

  buffer payload
    = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);

  struct bmp_data_node *data, *data_next;
  WALK_LIST_DELSAFE(data, data_next, p->update_msg_queue)
  {
    bmp_route_monitor_msg_serialize(&payload,
      data->global_peer, data->policy,
      data->remote_as, data->remote_id, true,
      data->remote_ip, data->data, data->data_size,
      data->timestamp);

    bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));

    bmp_buffer_flush(&payload);

    mb_free(data->data);
    rem_node(&data->n);
    mb_free(data);
  }

  bmp_buffer_free(&payload);
}

static void
bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs)
{
  TRACE(D_PACKETS, "Sending END-OF-RIB for %s.%s", ea_get_adata(bs->bgp, &ea_name)->data, ea_get_adata(bs->sender, &ea_name)->data);

  byte rx_end_payload[DEFAULT_MEM_BLOCK_SIZE];
  byte *pos = bgp_create_end_mark_ea_(bs->sender, rx_end_payload + BGP_HEADER_LENGTH);
  memset(rx_end_payload + BGP_MSG_HDR_MARKER_POS, 0xff,
	 BGP_MSG_HDR_MARKER_SIZE); // BGP UPDATE MSG marker
  put_u16(rx_end_payload + BGP_MSG_HDR_LENGTH_POS, pos - rx_end_payload);
  put_u8(rx_end_payload + BGP_MSG_HDR_TYPE_POS, PKT_UPDATE);

  bmp_route_monitor_put_update(p, bs, rx_end_payload, pos - rx_end_payload, current_real_time());
}

static void
bmp_send_peer_down_notif_msg(struct bmp_proto *p, ea_list *bgp,
  const byte *data, const size_t data_size)
{
  ASSERT(p->started);

  const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps_ext(bgp);
  bool is_global_instance_peer = bmp_is_peer_global_instance(bgp);
  buffer payload
    = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
  bmp_peer_down_notif_msg_serialize(&payload, is_global_instance_peer,
    ea_get_int(bgp, &ea_bgp_rem_as, 0), ea_get_int(bgp, &ea_bgp_rem_id, 0),
    remote_caps ? remote_caps->as4_support : ea_get_int(bgp, &ea_bgp_as4_session, 0),
    *((ip_addr*)ea_get_adata(bgp, &ea_bgp_rem_ip)->data), data, data_size);
  bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));

  bmp_buffer_free(&payload);
}

static void
bmp_peer_down_(struct bmp_proto *p, ea_list *bgp,
	       int err_class, int err_code, int err_subcode, const byte *data, int length)
{
  if (!p->started)
    return;

  struct bmp_peer *bp = bmp_find_peer(p, bgp);
  if (!bp)
    return;

  TRACE(D_STATES, "Peer down for %s", ea_find(bgp, &ea_name)->u.ad->data);

  uint bmp_code = 0;
  uint fsm_code = 0;

  switch (err_class)
  {
  case BE_BGP_RX:
    bmp_code = BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION;
    break;

  case BE_BGP_TX:
  case BE_AUTO_DOWN:
  case BE_MAN_DOWN:
    bmp_code = BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION;
    break;

  default:
    bmp_code = BMP_PEER_DOWN_REASON_REMOTE_NO_NOTIFICATION;
    length = 0;
    break;
  }

  buffer payload = bmp_buffer_alloc(p->buffer_mpool, 1 + BGP_HEADER_LENGTH + 2 + length);
  bmp_put_u8(&payload, bmp_code);

  switch (bmp_code)
  {
  case BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION:
  case BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION:
    bmp_put_bgp_hdr(&payload, BGP_HEADER_LENGTH + 2 + length, PKT_NOTIFICATION);
    bmp_put_u8(&payload, err_code);
    bmp_put_u8(&payload, err_subcode);
    bmp_put_data(&payload, data, length);
    break;

  case BMP_PEER_DOWN_REASON_LOCAL_NO_NOTIFICATION:
    bmp_put_u16(&payload, fsm_code);
    break;
  }

  bmp_send_peer_down_notif_msg(p, bgp, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));

  bmp_buffer_free(&payload);

  bmp_remove_peer(p, bp);
}

//#if 0
//void
//bmp_peer_down(const struct bgp_proto *bgp,
//	      int err_class, int code, int subcode, const byte *data, int length)
//{
//  int need_unlock = 1;
//  if (DG_IS_LOCKED(p->p.pool->domain))
//    need_unlock = 0;
//  else
//    DG_LOCK(p->p.pool->domain);

//  struct bmp_proto *p; node *n;
//  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
//    bmp_peer_down_(p, bgp, err_class, code, subcode, data, length);

//  if (need_unlock)
//    DG_UNLOCK(p->p.pool->domain);
//}
//#endif

static void
bmp_send_termination_msg(struct bmp_proto *p,
  const enum bmp_term_reason reason)
{
  const size_t term_msg_hdr_size = BMP_TERM_INFO_TYPE_SIZE
                                     + BMP_TERM_INFO_LEN_FIELD_SIZE
                                     + BMP_TERM_REASON_CODE_SIZE;
  const size_t term_msg_size = BMP_COMMON_HDR_SIZE + term_msg_hdr_size;
  buffer stream = bmp_buffer_alloc(p->buffer_mpool, term_msg_size);
  bmp_common_hdr_serialize(&stream, BMP_TERM_MSG, term_msg_hdr_size);
  bmp_put_u16(&stream, BMP_TERM_INFO_REASON);
  bmp_put_u16(&stream, BMP_TERM_REASON_CODE_SIZE); // 2-byte code indication the reason
  bmp_put_u16(&stream, reason);
  memcpy(p->sk->tbuf, bmp_buffer_data(&stream), bmp_buffer_pos(&stream));
  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
    sk_send(p->sk, bmp_buffer_pos(&stream)) < 0,
    "Failed to send BMP termination message"
    );

  log("bmp_send_termination_msg free %x", stream);
  bmp_buffer_free(&stream);
}

int
bmp_preexport(struct channel *C UNUSED, rte *e)
{
  /* Reject non-direct routes. Check if sender proto is the same as proto which created the route.
   * It assumes that route was created in a protocol.
   */
  struct rt_import_request *req = e->sender->req;
  struct channel *ch = SKIP_BACK(struct channel, in_req, req);

  struct rte_owner *owner = e->src->owner;
  struct proto *p = SKIP_BACK(struct proto, sources, owner);
  if (ch->proto != p)
    return -1;

  /* Reject non-BGP routes */
  if (p->proto != &proto_bgp)
    return -1;

  return 1;
}

int
bgp_next_hop_present(const rte *n)
{
  for (int i = 0; i < n->attrs->count; i++)
  {
    eattr a = n->attrs->attrs[i];
    const struct ea_class *class = ea_class_find(a.id);
    log("ea class %s", class->name);
    if (class == bgp_next_hop_ea_class)
      return 1;
  }
  log("BAD RTE, BAD.");
  return 0;
}

static void
solve_for_post_and_pre(struct bmp_proto *p, const rte *new, const rte *old)
{
  rte loc = *(new ?: old);

  struct proto *rte_proto = (struct proto*) SKIP_BACK(struct proto, sources, loc.src->owner);
  struct bgp_proto *bgp = (struct bgp_proto *) rte_proto;
  struct bgp_channel *src_ch = SKIP_BACK(struct bgp_channel, c.in_req, loc.sender->req);
  ASSERT_DIE(src_ch->c.proto == rte_proto); /* No pipes supported for now */

  log("solve_for_post_and_pre %s", rte_proto->name);

  /* Ignore non-BGP routes */
  if (rte_proto->proto != &proto_bgp)
    return;

  /* Checking the pre policy */
  if (p->monitoring_rib.in_pre_policy)
  {
    /* Compute the pre policy attributes */
    ea_list *new_attrs = new ? ea_strip_to(new->attrs, BIT32_ALL(EALS_PREIMPORT)) : NULL;
    ea_list *old_attrs = old ? ea_strip_to(old->attrs, BIT32_ALL(EALS_PREIMPORT)) : NULL;
    
    loc.attrs = new_attrs;

    if (new_attrs != old_attrs)
    {
      /* The attributes are actually different, announce the change */
      if (ea_same(new_attrs, old_attrs))
	bug("Two attribute sets are same in the attribute cache.");

      struct bmp_stream *bs = bmp_find_stream(p, bgp, src_ch->afi, false);
      if (bs)
      {
	log("s true bmp stream found feed, pre policy %i proto %s", bs->in_pre_policy, p->p.name);
	if (bmp_find_peer(p, proto_state_table->attrs[bgp->p.id]) == NULL)
	  bug("not implemented");

	bmp_route_monitor_notify(p, bgp, bs, &loc);
      }
    }
  }

  /* Checking the post policy */
  if (p->monitoring_rib.in_post_policy)
  {
    /* Compute the post policy attributes */
    ea_list *new_attrs = new ? ea_normalize(new->attrs, 0) : NULL;
    ea_list *old_attrs = old ? ea_normalize(old->attrs, 0) : NULL;

    loc.attrs = new_attrs;

    /* TODO: filter only BGP-relevant attributes */

    if ((new_attrs != old_attrs) || ea_same(new_attrs, old_attrs))
    {
      /* The attributes are actually different, announce the change */
      struct bmp_stream *bs = bmp_find_stream(p, bgp, src_ch->afi, true);
      if (bs)
      {
	log("s true bmp stream found feed, pre policy %i proto %s", bs->in_pre_policy, p->p.name);
	if (bmp_find_peer(p, proto_state_table->attrs[bgp->p.id]) == NULL)
	  bug("not implemented");

	bmp_route_monitor_notify(p, bgp, bs, &loc);
      }
    }
  }
}

static void
bmp_check_routes(void *bt_)
{
  log("bmp_check_routes");
  struct bmp_table *bt = (struct bmp_table *)bt_;
  struct bmp_proto *p = bt->p;

  RT_EXPORT_WALK(&bt->out_req, u) //const struct rt_export_union *_u;
  {
    log("feeder %x rte ptr %x", &bt->out_req, u);
    switch (u->kind)
    {
      case RT_EXPORT_STOP:
      	bug("Main table export stopped");

      case RT_EXPORT_FEED:
        log("export feed");
        uint oldpos = 0;
	      while ((oldpos < u->feed->count_routes) && !(u->feed->block[oldpos].flags & REF_OBSOLETE))
	        oldpos++;

	      /* Send updates one after another */
	      log("oldpos %i", oldpos);
	      for (uint i = 0; i < oldpos; i++)
	      {
	        rte *new = &u->feed->block[i];
	        log("pre policy  %i rte attr %x", p->monitoring_rib.in_pre_policy, new->attrs);

	        solve_for_post_and_pre(p, new, NULL);
	      }
	      break;
	    case RT_EXPORT_UPDATE:
	      log("export update");

	      solve_for_post_and_pre(p, u->update->new, u->update->old);
	      break;
	  }
  }
  log("end of notify fce");
}

static void
bmp_feed_end(struct rt_export_request *req)
{
  SKIP_BACK_DECLARE(struct bmp_table, bt, out_req, req);

  struct bmp_proto *p = bt->p;

log("bmp table found");
  /*
   * Unsynced streams are added in one moment during BMP session establishment,
   * therefore we can assume that all unsynced streams (for given channel)
   * already received full feed now and are synced.
   *
   * TODO: Use more efficent way to find bmp_stream from bmp_table
   */

  HASH_WALK(p->stream_map, next, bs)
  {
    if ((bs->table == bt) && !bs->sync)
    {
      bmp_route_monitor_end_of_rib(p, bs);
      bs->sync = true;
    }
  }
  HASH_WALK_END;
}


/**
 * bmp_startup - enter established state
 * @p: BMP instance
 *
 * The bgp_startup() function is called when the BMP session is established.
 * It sends initiation and peer up messagages.
 */
static void
bmp_startup(struct bmp_proto *p)
{
  ASSERT(!p->started);
  p->started = true;
  p->sock_err = 0;

  TRACE(D_EVENTS, "BMP session established");

  log("bmp startup %s %i", p->p.name, p->p.id);
  proto_notify_state(&p->p, PS_UP);

  /* Send initiation message */
  buffer payload = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
  bmp_init_msg_serialize(&payload, p->sys_descr, p->sys_name);
  bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
  bmp_buffer_free(&payload);

  /* Send Peer Up messages */
  for (u32 i = 0; i < proto_state_table->length; i++)
  {
    ea_list *proto_attr = proto_state_table->attrs[i];
    log("startup proto attr %i", proto_attr);
    if (proto_attr == NULL)
      continue;
    struct protocol *proto = (struct protocol *) ea_get_ptr(proto_attr, &ea_protocol_type, 0);
    const int state = ea_get_int(proto_attr, &ea_state, 0);
    log("proto bgp and up %i %i", proto != &proto_bgp, state != PS_UP);
    if (proto != &proto_bgp || state != PS_UP)
      continue;
    bmp_peer_init(p, proto_attr);
  }
  /*struct proto *peer;
  WALK_LIST(peer, proto_list)
    if ((peer->proto != &proto_bgp) && (peer->proto_state == PS_UP))
      bmp_peer_init(p, (struct bgp_proto *) peer);*/
}

/**
 * bmp_down - leave established state
 * @p: BMP instance
 *
 * The bgp_down() function is called when the BMP session fails. The caller is
 * responsible for changing protocol state.
 */
static void
bmp_down(struct bmp_proto *p)
{
  ASSERT(p->started);
  p->started = false;
  log("_!p->peer_map.count (%i) && !p->stream_map.count (%i)&& !p->table_map.count(%i)", p->peer_map.count, p->stream_map.count, p->table_map.count);

  TRACE(D_EVENTS, "BMP session closed");

  /* Unregister existing peer structures */
  HASH_WALK_DELSAFE(p->peer_map, next, bp)
  {
    log("bmp_remove_peer(p %x, bp %x)", p, bp);
    bmp_remove_peer(p, bp);
  }
  HASH_WALK_END;

  /* Removing peers should also remove all streams and tables */ //TODO
  log("!p->peer_map.count (%i) && !p->stream_map.count (%i)&& !p->table_map.count(%i)", p->peer_map.count, p->stream_map.count, p->table_map.count);
  ASSERT(!p->peer_map.count && !p->stream_map.count && !p->table_map.count);
}

/**
 * bmp_connect - initiate an outgoing connection
 * @p: BMP instance
 *
 * The bmp_connect() function creates the socket and initiates an outgoing TCP
 * connection to the monitoring station. It is called to enter Connect state.
 */
static void
bmp_connect(struct bmp_proto *p)
{
  ASSERT(!p->started);

  sock *sk = sk_new(p->p.pool);
  sk->type = SK_TCP_ACTIVE;
  sk->saddr = p->local_addr;
  sk->daddr = p->station_ip;
  sk->dport = p->station_port;
  sk->ttl = IP4_MAX_TTL;
  sk->tos = IP_PREC_INTERNET_CONTROL;
  sk->tbsize = BGP_TX_BUFFER_EXT_SIZE;
  sk->tx_hook = bmp_connected;
  sk->err_hook = bmp_sock_err;

  p->sk = sk;
  sk->data = p;

  TRACE(D_EVENTS, "Connecting to %I port %u", sk->daddr, sk->dport);

  int rc = sk_open(sk, p->p.loop);

  if (rc < 0)
    sk_log_error(sk, p->p.name);

  tm_start(p->connect_retry_timer, CONNECT_RETRY_TIME);
}

/* BMP connect successful event - switch from Connect to Established state */
static void
bmp_connected(struct birdsock *sk)
{
  struct bmp_proto *p = (void *) sk->data;

  TRACE(D_EVENTS, "Connected");

  sk->rx_hook = bmp_rx;
  sk->tx_hook = bmp_tx;
  tm_stop(p->connect_retry_timer);

  bmp_startup(p);
}

/* BMP socket error event - switch from any state to Idle state */
static void
bmp_sock_err(sock *sk, int err)
{
  struct bmp_proto *p = sk->data;

  p->sock_err = err;

  if (err)
    TRACE(D_EVENTS, "Connection lost (%M)", err);
  else
    TRACE(D_EVENTS, "Connection closed");

  if (p->started)
    bmp_down(p);

  bmp_close_socket(p);
  tm_start(p->connect_retry_timer, CONNECT_RETRY_TIME);

  proto_notify_state(&p->p, PS_START);
}

/* BMP connect timeout event - switch from Idle/Connect state to Connect state */
static void
bmp_connection_retry(timer *t)
{
  struct bmp_proto *p = t->data;

  if (p->started)
    return;

  bmp_close_socket(p);
  bmp_connect(p);
}

static void
bmp_close_socket(struct bmp_proto *p)
{
  rfree(p->sk);
  p->sk = NULL;
}


static void
bmp_postconfig(struct proto_config *CF)
{
  struct bmp_config *cf = (void *) CF;

  /* Do not check templates at all */
  if (cf->c.class == SYM_TEMPLATE)
    return;

  if (ipa_zero(cf->station_ip))
    cf_error("Station IP address not specified");

  if (!cf->station_port)
    cf_error("Station port number not specified");
}

void
fc_for_bmp_recipient(void *_p)
{
  struct bmp_proto *p = _p;

  ASSERT_DIE(birdloop_inside(p->p.loop));

  //log("received update, locked %i", locking_stack.service);
  struct lfjour_item *last_up;
  struct proto_pending_update *pupdate;
  while (last_up = lfjour_get(&p->proto_state_reader))
  {
    pupdate = SKIP_BACK(struct proto_pending_update, li, last_up);
    const byte *tx_open_msg = ea_get_adata(pupdate->proto_attr, &ea_bgp_local_open_msg)->data;

    int id = ea_get_int(pupdate->proto_attr, &ea_proto_id, 0);
    //log("id %i, tx = %i, l %i", id, tx_open_msg, ea_get_int(pupdate->proto_attr, &ea_bgp_local_open_msg_len, 0)); //FIXME why WHY are bmp eattrs able to give not null pointer?
    if (ea_get_int(pupdate->proto_attr, &ea_bgp_local_open_msg_len, 0))
    {
      struct protocol *proto = (struct protocol *) ea_get_ptr(pupdate->proto_attr, &ea_protocol_type, 0);
      //if (proto != &proto_bgp)
      //{
        //log("protocol %s is bgp %i", proto->name, proto == &proto_bgp);
        //bug("wrong protocol");
        //}
      //int id = ea_get_int(pupdate->proto_attr, &ea_proto_id, 0);
      const byte *rx_open_msg = ea_get_adata(pupdate->proto_attr, &ea_bgp_remote_open_msg)->data;
      int l_len = ea_get_int(pupdate->proto_attr, &ea_bgp_remote_open_msg_len, 0);
      int r_len = ea_get_int(pupdate->proto_attr, &ea_bgp_remote_open_msg_len, 0);
      bmp_peer_up_(p, proto_state_table->attrs[id], true, tx_open_msg, l_len, rx_open_msg, r_len);

    }
    else if (ea_get_int(pupdate->proto_attr, &ea_bgp_close_bmp_set, 0))
    {
      struct closing_bgp *closing = (struct closing_bgp *) ea_get_ptr(pupdate->proto_attr, &ea_protocol_type, 0);
      bmp_peer_down_(p, proto_state_table->attrs[id],
	       closing->err_class, closing->err_code, closing->err_subcode, closing->data, closing->length);
    }
	    
	    
    lfjour_release(&p->proto_state_reader, last_up);
  }
}

void
create_bmp_recipient(struct bmp_proto *p)
{
  struct lfjour_recipient *r = &p->proto_state_reader;
  r->event = &p->proto_state_changed;
  *r->event = (event) { .hook = fc_for_bmp_recipient, .data = p };
  log("p->p.loop %x", p->p.loop);
  r->target = birdloop_event_list(p->p.loop);
  
  LOCK_DOMAIN(rtable, proto_journal_domain);
  lfjour_register(proto_journal, r);
  UNLOCK_DOMAIN(rtable, proto_journal_domain);
  p->lf_jour_inited = 1;
}

/** Configuration handle section **/
static struct proto *
bmp_init(struct proto_config *CF)
{
  log("start init locked %i", locking_stack.service);
  struct proto *P = proto_new(CF);
  struct bmp_proto *p = (void *) P;
  struct bmp_config *cf = (void *) CF;

  if (!bgp_next_hop_ea_class)
    bgp_next_hop_ea_class = ea_class_find_by_name("bgp_next_hop");

  p->cf = cf;
  p->local_addr = cf->local_addr;
  p->station_ip = cf->station_ip;
  p->station_port = cf->station_port;
  strcpy(p->sys_descr, cf->sys_descr);
  strcpy(p->sys_name, cf->sys_name);
  p->monitoring_rib.in_pre_policy = cf->monitoring_rib_in_pre_policy;
  p->monitoring_rib.in_post_policy = cf->monitoring_rib_in_post_policy;
  //create_bmp_recipient(p); //should be here, but.
  log("new proto created locked %i", locking_stack.service);

  return P;
}

/**
 * bmp_start - initialize internal resources of BMP implementation.
 * NOTE: It does not connect to BMP collector yet.
 */
static int
bmp_start(struct proto *P)
{
  log("start locked %i", locking_stack.service);
  struct bmp_proto *p = (void *) P;

  p->buffer_mpool = rp_new(P->pool, proto_domain(&p->p), "BMP Buffer");
  p->map_mem_pool = rp_new(P->pool, proto_domain(&p->p), "BMP Map");
  p->tx_mem_pool = rp_new(P->pool, proto_domain(&p->p), "BMP Tx");
  p->update_msg_mem_pool = rp_new(P->pool, proto_domain(&p->p), "BMP Update");
  p->tx_ev = ev_new_init(p->p.pool, bmp_fire_tx, p);
  p->update_ev = ev_new_init(p->p.pool, bmp_route_monitor_commit, p);
  p->connect_retry_timer = tm_new_init(p->p.pool, bmp_connection_retry, p, 0, 0);
  p->sk = NULL;

  HASH_INIT(p->peer_map, P->pool, 4);
  HASH_INIT(p->stream_map, P->pool, 4);
  HASH_INIT(p->table_map, P->pool, 4);

  init_list(&p->tx_queue);
  init_list(&p->update_msg_queue);
  p->started = false;
  p->sock_err = 0;

  tm_start(p->connect_retry_timer, CONNECT_INIT_TIME);

  log("end of start locked %i", locking_stack.service);
  if (p->lf_jour_inited == 0)
    create_bmp_recipient(p);
  return PS_START;
}

static int
bmp_shutdown(struct proto *P)
{
  struct bmp_proto *p = (void *) P;

  if (p->started)
  {
    bmp_send_termination_msg(p, BMP_TERM_REASON_ADM);
    bmp_down(p);
  }

  p->sock_err = 0;

  return PS_DOWN;
}

static int
bmp_reconfigure(struct proto *P, struct proto_config *CF)
{
  struct bmp_proto *p = (void *) P;
  const struct bmp_config *new = (void *) CF;
  const struct bmp_config *old = p->cf;

  int needs_restart = bstrcmp(new->sys_descr, old->sys_descr)
    || bstrcmp(new->sys_name, old->sys_name)
    || !ipa_equal(new->local_addr, old->local_addr)
    || !ipa_equal(new->station_ip, old->station_ip)
    || (new->station_port != old->station_port)
    || (new->monitoring_rib_in_pre_policy != old->monitoring_rib_in_pre_policy)
    || (new->monitoring_rib_in_post_policy != old->monitoring_rib_in_post_policy);

  /* If there is any change, restart the protocol */
  if (needs_restart)
    return 0;

  /* We must update our copy of configuration ptr */
  p->cf = new;

  return 1;
}

static void
bmp_get_status(struct proto *P, byte *buf)
{
  struct bmp_proto *p = (void *) P;

  if (P->proto_state == PS_DOWN)
    bsprintf(buf, "Down");
  else
  {
    const char *state = !p->started ? (!p->sk ? "Idle" : "Connect") : "Established";

    if (!p->sock_err)
      bsprintf(buf, "%s", state);
    else
      bsprintf(buf, "%-14s%s %M", state, "Error:", p->sock_err);
  }
}

static void
bmp_show_proto_info(struct proto *P)
{
  struct bmp_proto *p = (void *) P;

  if (P->proto_state != PS_DOWN)
  {
    cli_msg(-1006, "  %-19s %I", "Station address:", p->station_ip);
    cli_msg(-1006, "  %-19s %u", "Station port:", p->station_port);

    if (!ipa_zero(p->local_addr))
      cli_msg(-1006, "  %-19s %I", "Local address:", p->local_addr);

    if (p->sock_err)
      cli_msg(-1006, "  %-19s %M", "Last error:", p->sock_err);
  }
}

struct protocol proto_bmp = {
  .name = "BMP",
  .template = "bmp%d",
  //.class = PROTOCOL_BMP, looks like there are no classes for protocols anymore
  .proto_size = sizeof(struct bmp_proto),
  .config_size = sizeof(struct bmp_config),
  .postconfig = bmp_postconfig,
  .init = bmp_init,
  .start = bmp_start,
  .shutdown = bmp_shutdown,
  .reconfigure = bmp_reconfigure,
  .get_status = bmp_get_status,
  .show_proto_info = bmp_show_proto_info,
};

void
bmp_build(void)
{
  proto_build(&proto_bmp);
}