sos_module_fetcher.c

嵌入式操作系统内核

/* -*- Mode: C; tab-width:4 -*- */
/* ex: set ts=4 shiftwidth=4 softtabstop=4 cindent: */
#include <hardware.h>
#include <codemem.h>
#include <sos_module_fetcher.h>
#include <sos_timer.h>
#include <message.h>
#include <sos_info.h>
#include <malloc.h>
#include <random.h>
#include <led_dbg.h>
#ifdef SOS_HAS_EXFLASH
#include <exflash.h>
#endif
#ifdef SOS_SIM
#include <sim_interface.h>
#endif
#include <led_dbg.h>

#ifndef SOS_DEBUG_FETCHER
#undef DEBUG
#define DEBUG(...)
#endif

/**
 * @brief A reliable transport service for modules
 *
 * The protocol will fetch data from one hop communication (i.e. no routing)
 * base on given module version, ID, and size of module.
 * When the protocol fetch is complete, it will inform the requester
 *
 * We do not queue any request at this moment.  The reason is that
 * the probability of getting more than one request is rather low.
 * This means that the components that use fetcher will have to handle the
 * failure
 */
static sos_module_t fetcher_module;

static inline void handle_overheard_fragment(Message *msg);
static int8_t handle_request(Message *msg);
static int8_t handle_data(Message *msg);
static void free_send_state_map(void);
static int8_t fetcher_handler(void *state, Message *msg);
static inline void handle_request_timeout(void);
static inline void send_fragment(void);
static inline void restart_request_timer(void);
static void check_map_and_post(void);
static void start_new_fetch(void);
static void send_fetcher_done(void);
static bool check_map(fetcher_bitmap_t *m);

#ifndef SOS_DEBUG_FETCHER
#define print_bitmap(m)
#else
static void print_bitmap(fetcher_bitmap_t *m);
#endif

static const mod_header_t mod_header SOS_MODULE_HEADER ={
  .mod_id = KER_FETCHER_PID,
  .state_size = 0,
  .num_timers = 1,
  .num_sub_func = 0,
  .num_prov_func = 0,
  .module_handler = fetcher_handler,
};

//! the status of sending fragment
enum {
    FETCHER_SENDING_FRAGMENT_INTERVAL  = 512L,
};

typedef struct {
    uint16_t dest;
    fetcher_fragment_t *fragr;  //!< the code fragment that will be stored
    fetcher_fragment_t *frag;  //!< the code fragment that will be sent
    fetcher_bitmap_t   *map;
    sos_timer_t        timer;
	uint8_t            num_funcs;
} fetcher_sending_state_t;

/**
 * @brief state for sending fragments
 * The reason we merge sending and receiving is because
 * we can send the fragment we have just received.
 */
static fetcher_sending_state_t send_state;

/**
 * @brief state for fetching fragments
 */
static fetcher_state_t *fst = NULL;

/**
 * @brief variable used for resending MSG_FETCHER_DONE
 * We use FETCHER_REQUEST_TID as retry timer
 */
static bool no_mem_retry = false;

int8_t fetcher_request(sos_pid_t req_id, sos_cam_t key, uint16_t size, uint16_t src)
{
    uint8_t bitmap_size;   //! size of the bitmap in bytes
    uint16_t num_fragments;
    uint8_t i;
    fetcher_state_t *f;
	fetcher_cam_t *cam;
	
	cam = (fetcher_cam_t *) ker_cam_lookup(key);
	if( cam == NULL ) return -EINVAL;

    //if(fst != NULL) return -EBUSY;
    DEBUG_PID(KER_FETCHER_PID, "fetcher_request, req_id = %d, size = %d, src = %d\n", req_id, size, src);

    num_fragments = ((size + (FETCHER_FRAGMENT_SIZE - 1))/ FETCHER_FRAGMENT_SIZE);
    bitmap_size = (uint8_t)((num_fragments + 7)/ 8);

    //DEBUG("size = %d\n", sizeof(fetcher_state_t) + bitmap_size);
    f = ker_malloc(sizeof(fetcher_state_t) + bitmap_size, KER_FETCHER_PID);
    if(f == NULL) {
        return -ENOMEM;
    }
    //DEBUG("num_fragments = %d, bitmap_zie = %d\n", num_fragments, bitmap_size);
    f->requester       = req_id;
    f->map.key         = key;
    f->map.bitmap_size = bitmap_size;
    f->src_addr        = src;
	f->num_funcs       = 0;
    f->next            = NULL;
	f->cm              = cam->cm;

    for(i = 0; i < bitmap_size; i++) {
        f->map.bitmap[i] = 0xff;
    }
    if((num_fragments) % 8) {
        f->map.bitmap[bitmap_size - 1] =
            (1 << (num_fragments % 8)) - 1;
    }
    print_bitmap(&f->map);

    //! backoff first!!!
    f->retx = 0;
    if(fst != NULL) {
        fetcher_state_t *tmp = fst;
		cam->status = FETCHING_QUEUED;

        while(tmp->next != NULL) { tmp = tmp->next; }
        tmp->next = f;
        return SOS_OK;
    }
	cam->status = FETCHING_STARTED;
    fst = f;
    //! setup timer
    ker_timer_start(KER_FETCHER_PID,
            FETCHER_REQUEST_TID,
            FETCHER_REQUEST_BACKOFF_SLOT * ((ker_rand() % FETCHER_REQUEST_MAX_SLOT) + 1));
    //DEBUG("request ret = %d\n", ret);
    return SOS_OK;
}

int8_t fetcher_cancel(sos_pid_t req_id, sos_cam_t key)
{
    fetcher_state_t *tmp;
    fetcher_state_t *prev;
    if( fst == NULL ) return -EINVAL;
    if( fst->map.key == key ) {	
        tmp = fst;
        start_new_fetch();
        ker_free( tmp );
		/*
		 * Cancel sender as well
		 */
		if( (send_state.map != NULL) && (send_state.map->key == key)) {
			free_send_state_map();
		}
        return SOS_OK;
    }
    prev = fst;
    tmp = fst->next;
    while(tmp != NULL) {
        if( tmp->map.key == key ) {	
            prev->next = tmp->next;
            ker_free( tmp );
            return SOS_OK;	
        }
        prev = tmp;
        tmp = tmp->next;
    }
    return -EINVAL;
}

void fetcher_restart(fetcher_state_t *s, uint16_t src)
{
	fetcher_cam_t *cam;
	cam = ker_cam_lookup( s->map.key );
    s->src_addr = src;
    s->retx = 0;
	s->next = NULL;
    if(fst != NULL) {
        fetcher_state_t *tmp = fst;
		cam->status = FETCHING_QUEUED;

        while(tmp->next != NULL) { tmp = tmp->next; }
        tmp->next = s;
        return;
    }
    fst = s;
	cam->status = FETCHING_STARTED;
    ker_timer_start(KER_FETCHER_PID,
            FETCHER_REQUEST_TID,
            FETCHER_REQUEST_BACKOFF_SLOT * ((ker_rand() % FETCHER_REQUEST_MAX_SLOT) + 1));
}

void fetcher_commit(fetcher_state_t *s, bool commit)
{
	fetcher_cam_t *cam;
	cam = (fetcher_cam_t *) ker_cam_lookup(s->map.key);
	if( cam == NULL ) return;
	if( commit == true ) {
		ker_codemem_flush( cam->cm, KER_FETCHER_PID );
	} else {
		ker_codemem_free( cam->cm );
	}
}

static void free_send_state_map()
{
	ker_free(send_state.map);
	send_state.map = NULL;
	ker_timer_stop(KER_FETCHER_PID, FETCHER_TRANSMIT_TID);
}


static int8_t fetcher_handler(void *state, Message *msg)
{
    switch (msg->type) {
        case MSG_FETCHER_FRAGMENT:
        {
			fetcher_fragment_t *f;
			f = (fetcher_fragment_t*)msg->data;
			f->key = entohs( f->key );
			f->frag_id = entohs(f->frag_id);

			DEBUG_PID(KER_FETCHER_PID,"MSG_FETCHER_FRAGMENT:\n");
			handle_overheard_fragment(msg);
			if(fst == NULL) {
				DEBUG_PID(KER_FETCHER_PID, "NO Request!!!\n");
				return SOS_OK;  //!< no request
			}
			//DEBUG_PID(KER_FETCHER_PID,"calling restart_request_timer()\n");
			restart_request_timer();
			fst->retx = 0;
			//DEBUG_PID(KER_FETCHER_PID,"calling handle_data()\n");
			return handle_data(msg);
		}
		case MSG_FETCHER_REQUEST:
		{
			fetcher_bitmap_t *bmap =
				(fetcher_bitmap_t *) msg->data;
			bmap->key = entohs( bmap->key );

			//! received request from neighbors
			DEBUG("handling request to %d from %d\n", msg->daddr, msg->saddr);
			if(msg->daddr == ker_id()) {
				return handle_request(msg);
			}
			if(fst == NULL) return SOS_OK;  //!< no request
			restart_request_timer();
			fst->retx = 0;
			return SOS_OK;
		}
		case MSG_TIMER_TIMEOUT:
		{
			MsgParam *params = (MsgParam*)(msg->data);
			if(params->byte == FETCHER_REQUEST_TID) {
				//DEBUG("request timeout\n");
				if( no_mem_retry ) {
					send_fetcher_done();
					return SOS_OK;
				}
				handle_request_timeout();
			} else if(params->byte == FETCHER_TRANSMIT_TID) {
				//DEBUG("send fragment timeout\n");
				send_fragment();
			}
			return SOS_OK;
		}
#ifdef SOS_HAS_EXFLASH
		case MSG_EXFLASH_WRITEDONE:
		{
			ker_free(send_state.fragr);
			send_state.fragr = NULL;
			check_map_and_post();
			return SOS_OK;
		}
		case MSG_EXFLASH_READDONE:
		{

			post_auto(KER_FETCHER_PID,
					KER_FETCHER_PID,
					MSG_FETCHER_FRAGMENT,
					sizeof(fetcher_fragment_t),
					send_state.frag,
					SOS_MSG_RELEASE,
					send_state.dest);
			send_state.frag = NULL;
			return SOS_OK;
		}
#endif
		case MSG_INIT:
		{
			send_state.map = NULL;
			send_state.frag = NULL;
			send_state.fragr = NULL;
			ker_msg_change_rules(KER_FETCHER_PID, SOS_MSG_RULES_PROMISCUOUS);
			ker_permanent_timer_init(&(send_state.timer), KER_FETCHER_PID, FETCHER_TRANSMIT_TID, TIMER_REPEAT);
			ker_timer_init(KER_FETCHER_PID, FETCHER_REQUEST_TID, TIMER_ONE_SHOT);
			return SOS_OK;
		}

	}
	return -EINVAL;
}

static inline void restart_request_timer()
{
	ker_timer_restart(KER_FETCHER_PID,
			FETCHER_REQUEST_TID,
			FETCHER_REQUEST_WATCHDOG + (FETCHER_REQUEST_BACKOFF_SLOT * ((ker_rand() % FETCHER_REQUEST_MAX_SLOT) + 1)));
}

static inline void handle_request_timeout()
{
	if(fst == NULL) {
		return;
	}
	//sos_assert(fst != NULL);
	//DEBUG("handle request timeout, retx = %d\n", fst->retx);
	fst->retx++;
	if(fst->retx <= FETCHER_REQUEST_MAX_RETX) {
		fetcher_bitmap_t *m;
		uint8_t size = sizeof(fetcher_bitmap_t) + fst->map.bitmap_size;

		DEBUG_PID(KER_FETCHER_PID,"send request to %d\n", fst->src_addr);
		print_bitmap(&fst->map);