routing_epidemic.c

Go to the documentation of this file.

/**
 * \addtogroup routing
 * @{
 */

/**
 * \defgroup routing_flooding Flooding Routing module
 *
 * @{
 */

/**
 * \file 
 * \brief implementation of flood routing
 * \author Georg von Zengen <vonzeng@ibr.cs.tu-bs.de>
 * \author Wolf-Bastian Poettner <poettner@ibr.cs.tu-bs.de>
 */

#include <string.h>

#include "net/netstack.h"
#include "net/rime/rimeaddr.h"
#include "lib/list.h"
#include "lib/memb.h"
#include "contiki.h"
#include "clock.h"
#include "logging.h"

#include "bundle.h"
#include "storage.h"
#include "sdnv.h"
#include "agent.h"
#include "discovery.h"
#include "statistics.h"
#include "bundleslot.h"
#include "delivery.h"
#include "convergence_layer.h"
#include "registration.h"

#include "routing.h"

/**
 * Internally used return values
 */
#define EPIDEMIC_ROUTE_RETURN_OK 1
#define EPIDEMIC_ROUTE_RETURN_CONTINUE 0
#define EPIDEMIC_ROUTE_RETURN_FAIL -1

struct neighbour_list_entry_t {
        /** address of the neighbour */
        rimeaddr_t neighbour;

        /** bundles that have been sent to the neighbour */
        uint32_t bundles[ROUTING_BUNDLES_PER_NEIGHBOUR];

        /** Point to where to *next* bundle number will be stored */
        uint8_t bundles_pointer;
};

struct routing_entry_t {
        /** Pointer to the next entry */
        struct routing_entry_t * next;

        /** number of the bundle */
        uint32_t bundle_number;

        /** bundle flags */
        uint8_t flags;

        /** destination node */
        uint32_t destination_node;
};

/**
 * Routing process
 */
PROCESS(routing_process, "Epidemic ROUTE process");

MEMB(routing_bundle_mem, struct routing_entry_t, BUNDLE_STORAGE_SIZE);
LIST(routing_bundle_list);

MEMB(routing_neighbour_mem, struct neighbour_list_entry_t, ROUTING_NEIGHBOURS);
LIST(routing_neigbour_list);

/**
 * \brief called by agent at startup
 */
void routing_epidemic_init(void)
{
        // Start CL process
        process_start(&routing_process, NULL);
}

/**
 * \brief Poll our process, so that we can resubmit bundles
 */
void routing_epidemic_schedule_resubmission(void)
{
        process_poll(&routing_process);
}

/**
 * \brief checks for a neighbour in the neighbour list
 * \param dest pointer to the address of the neighbor
 * \return pointer to the address list entry, NULL otherwise
 */
struct neighbour_list_entry_t * routing_epidemic_get_neighbour(rimeaddr_t * dest)
{
        struct neighbour_list_entry_t * entry = NULL;

        for(entry = list_head(routing_neigbour_list);
                entry != NULL;
                entry = list_item_next(entry)) {
                if( rimeaddr_cmp(dest, &entry->neighbour) ) {
                        /* Neighbour is in our list already, no need to store */
                        return entry;
                }
        }

        return NULL;
}

struct neighbour_list_entry_t * routing_epidemic_create_neighbour(rimeaddr_t * dest)
{
        struct neighbour_list_entry_t * entry = NULL;

        entry = routing_epidemic_get_neighbour(dest);
        if( entry != NULL ) {
                return entry;
        }

        /* allocate neighbour list entry */
        entry = memb_alloc(&routing_neighbour_mem);
        if( entry == NULL ) {
                return NULL;
        }

        /* Initialize struct */
        memset(entry, 0, sizeof(struct neighbour_list_entry_t));

        /* fill in the data */
        rimeaddr_copy(&entry->neighbour, dest);

        /* add the struct to the list */
        list_add(routing_neigbour_list, entry);

        return entry;
}

void routing_epidemic_delete_neighbour(rimeaddr_t * dest)
{
        struct neighbour_list_entry_t * entry = NULL;

        entry = routing_epidemic_get_neighbour(dest);
        if( entry != NULL ) {
                return;
        }

        /* Remove entry from list */
        list_remove(routing_neigbour_list, entry);

        /* Free entry */
        memb_free(&routing_neighbour_mem, entry);
}

void routing_epidemic_store_sent_bundle(rimeaddr_t * dest, uint32_t bundle_number)
{
        struct neighbour_list_entry_t * n = NULL;
        int h;

        /* Get the neighbour to which we have delivered the bundle */
        n = routing_epidemic_create_neighbour(dest);

        /* Look, if we have sent this bundle to the neighbour earlier */
        for(h=0; h<n->bundles_pointer; h++) {
                if( n->bundles[h] == bundle_number ) {
                        return;
                }
        }

        /* Note down bundle in list of neighbour */
        n->bundles[n->bundles_pointer] = bundle_number;
        n->bundles_pointer ++;
}

uint8_t routing_epidemic_was_bundle_sent(rimeaddr_t * dest, uint32_t bundle_number)
{
        struct neighbour_list_entry_t * n = NULL;
        int h;

        /* Get the neighbour to which we have delivered the bundle */
        n = routing_epidemic_create_neighbour(dest);

        /* Look, if we have sent this bundle to the neighbour earlier */
        for(h=0; h<n->bundles_pointer; h++) {
                if( n->bundles[h] == bundle_number ) {
                        return 1;
                }
        }

        return 0;
}

/**
 * \brief checks if there are bundle to send to dest
 * \param dest pointer to the address of the new neighbor
 */
void routing_epidemic_new_neighbour(rimeaddr_t *dest)
{
        routing_epidemic_create_neighbour(dest);
}


/**
 * \brief Send bundle to neighbour
 * \param bundle_number Number of the bundle
 * \param neighbour Address of the neighbour
 * \return 1 on success, -1 on error
 */
int routing_epidemic_send_bundle(uint32_t bundle_number, rimeaddr_t * neighbour)
{
        struct transmit_ticket_t * ticket = NULL;

        /* Allocate a transmission ticket */
        ticket = convergence_layer_get_transmit_ticket();
        if( ticket == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_WRN, "unable to allocate transmit ticket");
                return -1;
        }

        /* Specify which bundle */
        rimeaddr_copy(&ticket->neighbour, neighbour);
        ticket->bundle_number = bundle_number;

        /* Put the bundle in the queue */
        convergence_layer_enqueue_bundle(ticket);

        return 1;
}

/**
 * \brief Deliver a bundle to a local service
 * \param entry Pointer to the routing entry of the bundle
 * \return EPIDEMIC_ROUTE_RETURN_OK if delivered, EPIDEMIC_ROUTE_RETURN_CONTINUE otherwise
 */
int routing_epidemic_send_to_local(struct routing_entry_t * entry)
{
        struct mmem * bundlemem = NULL;

        // Should this bundle be delivered locally?
        if( (entry->flags & ROUTING_FLAG_LOCAL) && !(entry->flags & ROUTING_FLAG_IN_DELIVERY) ) {
                bundlemem = BUNDLE_STORAGE.read_bundle(entry->bundle_number);
                if( bundlemem == NULL ) {
                        LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "cannot read bundle %lu", entry->bundle_number);
                        return EPIDEMIC_ROUTE_RETURN_CONTINUE;
                }

                if( delivery_deliver_bundle(bundlemem) ) {
                        entry->flags |= ROUTING_FLAG_IN_DELIVERY;
                }
        }

        return EPIDEMIC_ROUTE_RETURN_OK;
}

/**
 * \brief Forward a bundle to the next hop
 * \param entry Pointer to the routing entry of the bundle
 * \return EPIDEMIC_ROUTE_RETURN_OK if queued, EPIDEMIC_ROUTE_RETURN_CONTINUE if not queued and EPIDEMIC_ROUTE_RETURN_FAIL of queue is full
 */
int routing_epidemic_forward(struct routing_entry_t * entry)
{
        struct neighbour_list_entry_t * n = NULL;
        int h = 0;

        for( n = list_head(routing_neigbour_list);
                 n != NULL;
                 n = list_item_next(n) ) {

                if( routing_epidemic_was_bundle_sent(&n->neighbour, entry->bundle_number) ) {
                        continue;
                }

                LOG(LOGD_DTN, LOG_ROUTE, LOGL_INF, "send bundle %lu to %u.%u", entry->bundle_number, n->neighbour.u8[0], n->neighbour.u8[1]);

                /* Mark bundle as busy */
                entry->flags |= ROUTING_FLAG_IN_TRANSIT;

                /* And queue it for sending */
                h = routing_epidemic_send_bundle(entry->bundle_number, &n->neighbour);
                if( h < 0 ) {
                        /* Enqueuing bundle failed - unblock it */
                        entry->flags &= ~ROUTING_FLAG_IN_TRANSIT;

                        /* If sending the bundle fails, all other will likely also fail */
                        return EPIDEMIC_ROUTE_RETURN_FAIL;
                }

                /* Only one bundle at a time */
                return EPIDEMIC_ROUTE_RETURN_OK;
        }

        return EPIDEMIC_ROUTE_RETURN_CONTINUE;
}

/**
 * \brief iterate through all bundles and forward bundles
 */
void routing_epidemic_send_to_known_neighbours(void)
{
        struct routing_entry_t * n = NULL;
        int try_to_forward = 1;
        int h = 0;

        LOG(LOGD_DTN, LOG_ROUTE, LOGL_DBG, "send to known neighbours");

        /**
         * It is likely, that we will have less neighbours than bundles - therefore, we want to to go through bundles only once
         */
        for( n = (struct routing_entry_t *) list_head(routing_bundle_list);
                 n != NULL;
                 n = list_item_next(n) ) {

                /* Is the bundle for local? */
                h = routing_epidemic_send_to_local(n);
                /* We do not care about the return value, because we would continue anyway */

                /* Skip this bundle, if it is not queued for forwarding */
                if( !(n->flags & ROUTING_FLAG_FORWARD) || (n->flags & ROUTING_FLAG_IN_TRANSIT) || !try_to_forward ) {
                        continue;
                }

                /* At this point, we know that the bundle is not for one of our neighbours, so send it to all the others */
                h = routing_epidemic_forward(n);
                if( h == EPIDEMIC_ROUTE_RETURN_OK ) {
                        /* Bundle will be forwarded, continue as normal */
                } else if( h == EPIDEMIC_ROUTE_RETURN_CONTINUE ) {
                        /* Bundle will not be forwarded, continue as normal */
                } else if( h == EPIDEMIC_ROUTE_RETURN_FAIL ) {
                        /* Enqueuing the bundle failed, to stop the forwarding process */
                        try_to_forward = 0;
                        continue;
                }
        }
}

/**
 * \brief Wrapper function for agent calls to resubmit bundles for already known neighbours
 */
void routing_epidemic_resubmit_bundles() {
        routing_epidemic_schedule_resubmission();
}

/**
 * \brief Adds a new bundle to the list of bundles
 * \param bundle_number bundle number of the bundle
 * \return >0 on success, <0 on error
 */
int routing_epidemic_new_bundle(uint32_t * bundle_number)
{
        struct routing_entry_t * n = NULL;
        struct mmem * bundlemem = NULL;
        struct bundle_t * bundle = NULL;
        uint32_t source_node_eid;
        rimeaddr_t source_node;

        LOG(LOGD_DTN, LOG_ROUTE, LOGL_DBG, "agent announces bundle %lu", *bundle_number);

        // Let us see, if we know this bundle already
        for( n = list_head(routing_bundle_list);
                 n != NULL;
                 n = list_item_next(n) ) {

                if( n->bundle_number == *bundle_number ) {
                        LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "agent announces bundle %lu that is already known", *bundle_number);
                        return -1;
                }
        }

        // Notify statistics
        statistics_bundle_incoming(1);

        // Now allocate new memory for the list entry
        n = memb_alloc(&routing_bundle_mem);
        if( n == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "cannot allocate list entry for bundle, please increase BUNDLE_STORAGE_SIZE");
                return -1;
        }

        /* Initialize struct */
        memset(n, 0, sizeof(struct routing_entry_t));

        // Now go and request the bundle from storage
        bundlemem = BUNDLE_STORAGE.read_bundle(*bundle_number);
        if( bundlemem == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "unable to read bundle %lu", *bundle_number);
                memb_free(&routing_bundle_mem, n);
                return -1;
        }

        // Get our bundle struct and check the pointer
        bundle = (struct bundle_t *) MMEM_PTR(bundlemem);
        if( bundle == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "invalid bundle pointer for bundle %lu", *bundle_number);
                memb_free(&routing_bundle_mem, n);
                bundle_decrement(bundlemem);
                return -1;
        }

        // Nothing can go wrong anymore, add the (surrounding) struct to the list
        list_add(routing_bundle_list, n);

        /* Here we decide if a bundle is to be delivered locally and/or forwarded */
        if( bundle->dst_node == dtn_node_id ) {
                /* This bundle is for our node_id, deliver locally */
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_DBG, "bundle is for local");
                n->flags |= ROUTING_FLAG_LOCAL;
        } else {
                /* This bundle is not (directly) for us and will be forwarded */
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_DBG, "bundle is for forward");
                n->flags |= ROUTING_FLAG_FORWARD;
        }

        if( !(bundle->flags & BUNDLE_FLAG_SINGLETON) ) {
                /* Bundle is not Singleton, so forward it in any case */
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_DBG, "bundle is for forward");
                n->flags |= ROUTING_FLAG_FORWARD;
        }

        if( registration_is_local(bundle->dst_srv, bundle->dst_node) && bundle->dst_node != dtn_node_id) {
                /* Bundle is for a local registration, so deliver it locally */
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_DBG, "bundle is for local and forward");
                n->flags |= ROUTING_FLAG_LOCAL;
                n->flags |= ROUTING_FLAG_FORWARD;
        }

        // Now copy the necessary attributes from the bundle
        n->bundle_number = *bundle_number;
        bundle_get_attr(bundlemem, DEST_NODE, &n->destination_node);

        // Note down, that we have received this bundle from a certain neighbour -> do not send it again
        bundle_get_attr(bundlemem, SRC_NODE, &source_node_eid);
        source_node = convert_eid_to_rime(source_node_eid);
        routing_epidemic_store_sent_bundle(&source_node, *bundle_number);
        routing_epidemic_store_sent_bundle(&bundle->msrc, *bundle_number);

        // Now that we have the bundle, we do not need the allocated memory anymore
        bundle_decrement(bundlemem);
        bundlemem = NULL;
        bundle = NULL;

        // Schedule to deliver and forward the bundle
        routing_epidemic_schedule_resubmission();

        // We do not have a failure here, so it must be a success
        return 1;
}

/**
 * \brief deletes bundle from list
 * \param bundle_number bundle number of the bundle
 */
void routing_epidemic_delete_bundle(uint32_t bundle_number)
{
        struct routing_entry_t * n = NULL;

        LOG(LOGD_DTN, LOG_ROUTE, LOGL_DBG, "EPIDEMIC_del_bundle for bundle %lu", bundle_number);

        // Find the bundle in our internal storage
        for( n = list_head(routing_bundle_list);
                 n != NULL;
                 n = list_item_next(n) ) {

                if( n->bundle_number == bundle_number ) {
                        break;
                }
        }

        if( n == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "epidemic_del_bundle for bundle %lu that we do not know", bundle_number);
                return;
        }

        list_remove(routing_bundle_list, n);

        // And also free the memory for the list entry
        memb_free(&routing_bundle_mem, n);
}

/**
 * \brief Callback function informing us about the status of a sent bundle
 * \param ticket CL transmit ticket of the bundle
 * \param status status code
 */
void routing_epidemic_bundle_sent(struct transmit_ticket_t * ticket, uint8_t status)
{
        struct routing_entry_t * n = NULL;

        // Tell the agent to call us again to resubmit bundles
        routing_epidemic_schedule_resubmission();

        // Find the bundle in our internal storage
        for( n = list_head(routing_bundle_list);
                 n != NULL;
                 n = list_item_next(n) ) {

                if( n->bundle_number == ticket->bundle_number ) {
                        break;
                }
        }

        if( n == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "Bundle not in storage");
                return;
        }

        /* Bundle is not busy anymore */
        n->flags &= ~ROUTING_FLAG_IN_TRANSIT;

        if( status == ROUTING_STATUS_NACK ||
                status == ROUTING_STATUS_FAIL ) {
                // NACK = Other side rejected the bundle, try again later
                // FAIL = Transmission failed
                // --> note down address in blacklist

                /* Free up the ticket */
                convergence_layer_free_transmit_ticket(ticket);

                return;
        }

        if( status == ROUTING_STATUS_ERROR ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "Bundle %lu has fatal error, deleting", ticket->bundle_number);

                /* Bundle failed permanently, we can delete it because it will never be delivered anyway */
                n->flags = 0;

                /* Free up the ticket */
                convergence_layer_free_transmit_ticket(ticket);

                return;
        }

        // Here: status == ROUTING_STATUS_OK
        statistics_bundle_outgoing(1);

        /* Note down, that we have sent the bundle to this neighbour */
        routing_epidemic_store_sent_bundle(&ticket->neighbour, ticket->bundle_number);

        /* Free up the ticket */
        convergence_layer_free_transmit_ticket(ticket);
}

/**
 * \brief Incoming notification, that service has finished processing bundle
 * \param bundlemem Pointer to the MMEM struct of the bundle
 */
void routing_epidemic_bundle_delivered_locally(struct mmem * bundlemem) {
        struct routing_entry_t * n = NULL;
        struct bundle_t * bundle = (struct bundle_t *) MMEM_PTR(bundlemem);

        // Tell the agent to call us again to resubmit bundles
        routing_epidemic_schedule_resubmission();

        if( bundle == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "EPIDEMIC_locally_delivered called with invalid pointer");
                return;
        }

        // Find the bundle in our internal storage
        for( n = (struct routing_entry_t *) list_head(routing_bundle_list);
                 n != NULL;
                 n = list_item_next(n) ) {

                if( n->bundle_number == bundle->bundle_num ) {
                        break;
                }
        }

        if( n == NULL ) {
                LOG(LOGD_DTN, LOG_ROUTE, LOGL_ERR, "Bundle not in storage yet");
                return;
        }

        // Unset the IN_DELIVERY flag
        n->flags &= ~ROUTING_FLAG_IN_DELIVERY;

        // Unset the LOCAL flag
        n->flags &= ~ROUTING_FLAG_LOCAL;

        // Unblock the receiving service
        delivery_unblock_service(bundlemem);

        // Free the bundle memory
        bundle_decrement(bundlemem);

        /* Note down, that we have sent the bundle to this neighbour */
        routing_epidemic_store_sent_bundle(&rimeaddr_node_addr, bundle->bundle_num);
}

/**
 * \brief Routing persistent process
 */
PROCESS_THREAD(routing_process, ev, data)
{
        PROCESS_BEGIN();

        LOG(LOGD_DTN, LOG_ROUTE, LOGL_INF, "FLOOD ROUTE process in running");

        /* Initialize memory used to store bundles for routing */
        memb_init(&routing_bundle_mem);
        list_init(routing_bundle_list);

        /* Initialize memory used to store neighbours */
        memb_init(&routing_neighbour_mem);
        list_init(routing_neigbour_list);

        while(1) {
                PROCESS_YIELD_UNTIL(ev == PROCESS_EVENT_POLL);

                routing_epidemic_send_to_known_neighbours();
        }

        PROCESS_END();
}

const struct routing_driver routing_epidemic ={
        "EPIDEMIC_route",
        routing_epidemic_init,
        routing_epidemic_new_neighbour,
        routing_epidemic_new_bundle,
        routing_epidemic_delete_bundle,
        routing_epidemic_bundle_sent,
        routing_epidemic_resubmit_bundles,
        routing_epidemic_bundle_delivered_locally,
};