vaddr_t
cbuf_c_register(spdid_t spdid, long cbid)
{
	struct spd_tmem_info *sti;
	vaddr_t p, mgr_addr;

	/* DOUT("\nREGISTERED!!!\n"); */
	sti = get_spd_info(spdid);
	
	mgr_addr = (vaddr_t)alloc_page();
	p = (vaddr_t)valloc_alloc(cos_spd_id(), spdid, 1);
	if (p !=
	    (mman_alias_page(cos_spd_id(), mgr_addr, spdid, p))) {
		DOUT("mapped faied p is %p\n",(void *)p);
		valloc_free(cos_spd_id(), spdid, (void *)p, 1);
		return -1;
	}
	sti->managed = 1;
	/* __spd_cbvect_add_range(sti, cbid, (struct cbuf_vect_intern_struct *)mgr_addr); */
	__spd_cbvect_add_range(sti, cbid, mgr_addr);

	return p;
}

int __sg_sched_block(spdid_t spdid, unsigned short int dependency_thd)
{
	/* printc("ser: sched_block (thd %d)\n", cos_get_thd_id()); */
	int ret;
#ifdef LOG_MONITOR
	monevt_enqueue(cos_spd_id(), 11, dependency_thd);
#endif
	ret = sched_block(spdid, dependency_thd);
#ifdef LOG_MONITOR
	monevt_enqueue(0, 11, dependency_thd);
#endif

	return ret;
}

int __sg_sched_component_take(spdid_t spdid)
{
	/* printc("ser: sched_component_take (thd %d)\n", cos_get_thd_id()); */
	int ret;
#ifdef LOG_MONITOR
	monevt_enqueue(cos_spd_id(), 13, 0);
#endif
	ret = sched_component_take(spdid);
#ifdef LOG_MONITOR
	monevt_enqueue(0, 13, 0);
#endif

	return ret;
}

int __sg_sched_wakeup(spdid_t spdid, unsigned short int thd_id)
{
	/* printc("ser: sched_wakeup (thd %d)\n", cos_get_thd_id()); */
	int ret;
#ifdef LOG_MONITOR
	monevt_enqueue(cos_spd_id(), 12, thd_id);
#endif
	ret = sched_wakeup(spdid, thd_id);
#ifdef LOG_MONITOR
	monevt_enqueue(0, 12, thd_id);
#endif

	return ret;
}

static void from_data_new(struct tor_conn *tc)
{
	int from, to, amnt;
	char *buf;

	from = tc->from;
	to   = tc->to;
	buf = cos_argreg_alloc(BUFF_SZ);
	assert(buf);
	while (1) {
		int ret;

		amnt = net_recv(cos_spd_id(), from, buf, BUFF_SZ-1);
		if (0 == amnt) break;
		else if (-EPIPE == amnt) {
			goto close;
		} else if (amnt < 0) {
			printc("read from fd %d produced %d.\n", from, amnt);
			BUG();
		}
		if (amnt != (ret = twrite_pack(cos_spd_id(), to, buf, amnt))) {
			printc("conn_mgr: write failed w/ %d on fd %d\n", ret, to);
			goto close;

		}
	}
done:
	cos_argreg_free(buf);
	return;
close:
	net_close(cos_spd_id(), from);
	trelease(cos_spd_id(), to);
	tor_del_pair(from, to);
	if (tc->feid) cvect_del(&evts, tc->feid);
	if (tc->teid) cvect_del(&evts, tc->teid);
	goto done;
}

void core1_high() {
	printc("core %ld high prio thd %d running.\n", cos_cpuid(), cos_get_thd_id());

	create_thd(0, HIGH_PRIO);

	create_thd(1, LOW_PRIO);

	/* Brand operations removed. Add acap creation here. */
	int received_ipi = 0;

	int param[4];
	u64_t s, e;
	int iter = 0;
	while (1) {
		int ret = 0;
		/* printc("core %ld going to wait, thd %d\n", cos_cpuid(), cos_get_thd_id()); */

		/* if (-1 == (ret = cos_ainv_wait(...))) BUG(); */

		/* printc("core %ld, rec %d\n", cos_cpuid(), ++received_ipi); */
		param[0] = shared_mem[0];
		param[1] = shared_mem[1];
		param[2] = shared_mem[2];
		param[3] = shared_mem[3];
		assert(param[0] == 2);
		assert(param[1] == 4);
		assert(param[2] == 6);
		assert(param[3] == 8);

		/* rdtscll(e); */
		/* data[iter++] = e - c1_tsc; */

		int i;
		for (i = 0; i < n_wait; i++) {
			delay(20);
			/* printc("core %d triggering evt %d, i %d....\n", cos_cpuid(), evt, i); */
			shared_ret = 10;

			/* rdtscll(s); */

			evt_trigger(cos_spd_id(), evt);
 
			/* rdtscll(e); */
			/* data[iter++] = e - s; */

			/* printc("core %d triggerred evt %d, i %d....\n", cos_cpuid(), evt, i); */
		}
	}
}

void cos_init(void)
{
	static int first = 0, flag = 0;
	union sched_param sp;

	if(first == 0){
		first = 1;
		sp.c.type = SCHEDP_PRIO;
		sp.c.value = 8;
		high = sched_create_thd(cos_spd_id(), sp.v, 0, 0);
	} else {
		if (cos_get_thd_id() == high) {
			periodic_wake_create(cos_spd_id(), NOISE_PERIOD);
			while(1){
				periodic_wake_wait(cos_spd_id());
				/* printc("PERIODIC: noise....(thd %d in spd %ld)\n", */
				/*        cos_get_thd_id(), cos_spd_id()); */
				sinusoid_spike();
			}
		}
	}
	
	return;
}

void
trelease(spdid_t spdid, td_t td)
{
	struct torrent *t;
	td_t ntd;

	if (!tor_is_usrdef(td)) return;
	t = tor_lookup(td);
	if (!t) goto done;
	ntd = (td_t)t->data;
	parent_trelease(cos_spd_id(), ntd);
	tor_free(t);
done:
	return;
}

static void 
accept_new(int accept_fd)
{
	int eid;

	eid = evt_get();
	assert(eid > 0);
	pid_torrent = from_tsplit(cos_spd_id(), accept_fd, "", 0, TOR_RW, eid);
	assert(pid_torrent!= accept_fd);
	
	printc("accept_new: eid %d pid_torrent %d (accept_fd %d)\n", 
	       eid, pid_torrent, accept_fd);
	
	if (-EAGAIN == pid_torrent) {
		evt_free(cos_spd_id(), eid);
		return;
	} else if (pid_torrent < 0) {
		printc("pwrite to id_torrent %d\n", pid_torrent);
		BUG();
		return;
	}
	
	mapping_add(pid_torrent, 0, eid, 0);
}

static void
revoke_test()
{
	int i;
	vaddr_t addr = 0;
	printc("\n<<< REVOKE TEST BEGIN! >>>\n");

#ifdef TEN2TEN  		/* 10 to 10 */
	for (i = 0; i<PAGE_NUM; i++) {
		addr = s_addr[i];
		/* printc("s_addr %p\n", addr); */
		/* rdtscll(start); */
		mman_revoke_page(cos_spd_id(), addr, 0);
		/* rdtscll(end); */
		/* printc("COST (mman_revoke_page) %llu\n", end - start); */
	}
#else  /* 1 to 10 */
	addr = s_addr[0];
	/* printc("s_addr %p\n", addr); */
	mman_revoke_page(cos_spd_id(), addr, 0);
#endif
	printc("<<< REVOKE TEST END! >>>\n\n");
	return;
}

/* restore the server state */
static void
rd_recover_state(struct rec_data_tor *rd)
{
    struct rec_data_tor *prd, *tmp = NULL;
    char val[10]; // 2^32 use 10 bits

    assert(rd && rd->p_tid >= 1 && rd->c_tid > 1);

    /* printc("in rd_recover_state: rd->p_tid %d\n", rd->p_tid); */
    if (rd->p_tid > 1) {     // not tsplit from td_root
        assert((prd = map_rd_lookup(rd->p_tid)));
        prd->fcnt = global_fault_cnt;
        /* printc("in rd_recover_state: found a parent to be recovered rd->p_tid %d\n", */
        /*        rd->p_tid); */
        rd_recover_state(prd);
    }

    // has reached td_root, start rebuilding and no tracking...
    // tsplit returns the client id !!!!
    /* printc("\n recovery process calls tsplit again!!!...\n\n"); */
    /* printc("saved param is %s\n", rd->param); */
    td_t tmp_tid = fs_tsplit(cos_spd_id(), rd->p_tid,
                             rd->param, rd->param_len, rd->tflags, rd->evtid);
    if (tmp_tid <= 1) return;
    /* printc("\nrecovery process tsplit return!!!...(tmp_tid %d)\n\n", tmp_tid); */

    assert((tmp = map_rd_lookup(tmp_tid)));
    rd->s_tid = tmp->s_tid;
    /* printc("got the new client side %d and its new server id %d\n",  */
    /*        tmp_tid, tmp->s_tid); */

    /* do not track the new tid for retsplitting.. (wish to avoid
    * this) add this to ramfs as well */
    map_rd_delete(tmp_tid);

    /* //Now bring the data back as well */
    /* printc("\nnow it is time to bring the data back...\n\n"); */

    /* rd->being_recovered = 1; */
    /* int ret = -1; */
    /* sprintf(val, "%d", rd->s_tid); */
    /* printc("val %s val_len %d (td %d)\n", val, strlen(val), rd->s_tid); */
    /* ret = twmeta(cos_spd_id(), rd->s_tid, "data", strlen("data"), val, strlen(val)); */
    /* assert(!ret); */

    /* printc("\nnow the data is brought back!!!!\n\n"); */
    return;
}

void *valloc_alloc(spdid_t spdid, spdid_t dest, unsigned long npages)
{
	/* JWW print out a few things : spdid, heap ptr, make sure the heap ptr is sane */

	void *ret = NULL;
	struct spd_vas_tracker *trac;
	struct spd_vas_occupied *occ;
	long off;
	/*JWW REMOVE THIS */
	struct cos_component_information *ci;
	unsigned long page_off;
	void *hp;
	/* /JWW */

	LOCK();
	/*JWW REMOVE THIS */
	ci = cos_get_vas_page();
	if (cinfo_map(cos_spd_id(), (vaddr_t)ci, spdid)) {
		// error
		cos_release_vas_page(ci);
		printc("CINFO_MAP ERROR\n");
	}
	hp = (void*)ci->cos_heap_ptr;
	// now print some things out.
	//	printc("valloc alloc heap_ptr: %x, ucap_tbl: %x, npages: %ul \n", (unsigned int) hp, (unsigned int) ci->cos_user_caps, npages);
	/* /JWW */

	page_off = ((unsigned long)hp - (unsigned long)round_to_pgd_page(hp))/PAGE_SIZE;

	trac = cos_vect_lookup(&spd_vect, dest);
	if (!trac) {
		printc("valloc init being called\n");
		if (__valloc_init(dest) ||
		    !(trac = cos_vect_lookup(&spd_vect, dest))) goto done;
	}
	//	printc("valloc alloc past init\n");
	
	occ = trac->map;
	assert(occ);
	//	off = bitmap_extent_find_set(&occ->pgd_occupied[0], page_off, npages, MAP_MAX);
	off = bitmap_extent_find_set(&occ->pgd_occupied[0], 0, npages, MAP_MAX);
	if (off < 0) goto done;
	ret = ((char *)trac->extents[0].start) + (off * PAGE_SIZE);
done:   
	//	printc("valloc alloc returning %x\n", (unsigned int) ret);
	UNLOCK();
	return ret;
}

int timed_event_wakeup(spdid_t spdinv, unsigned short int thd_id)
{
	spdid_t spdid = cos_spd_id();
	struct thread_event *evt;

	TAKE(spdid);
	ticks = sched_timestamp();
	if (NULL == (evt = find_remove_event(thd_id))) {
		RELEASE(spdid);
		return 1;
	}
	RELEASE(spdid);
	assert(evt->thread_id == thd_id);

	return sched_wakeup(spdid, thd_id);
}

static inline void *
__page_get(void)
{
	void *hp = cos_get_vas_page();
	struct frame *f = frame_alloc();

	assert(hp && f);
	frame_ref(f);
	f->nmaps  = -1; 	 /* belongs to us... */
	f->c.addr = (vaddr_t)hp; /* ...at this address */
	if (cos_mmap_cntl(COS_MMAP_GRANT, MAPPING_RW, cos_spd_id(), (vaddr_t)hp, frame_index(f))) {
		printc("grant @ %p for frame %d\n", hp, frame_index(f));
		BUG();
	}
	return hp;
}

void 
cos_init(void)
{
	static int first = 0;
	union sched_param sp;
	if(first == 0){
		first = 1;

		sp.c.type = SCHEDP_PRIO;
		sp.c.value = 11;
		high = sched_create_thd(cos_spd_id(), sp.v, 0, 0);
	} else {
		test0();
	}
	return;
}

/* As above, but return more than one event notifications */
int evt_grp_mult_wait(spdid_t spdid, struct cos_array *data)
{
	struct evt_grp *g;
	struct evt *e = NULL;
	int evt_gathered = 0, evt_max;

	if (!cos_argreg_arr_intern(data)) return -EINVAL;
	evt_max = data->sz / sizeof(long);

	while (1) {
		lock_take(&evt_lock);

		g = evt_grp_find(cos_get_thd_id());
		ACT_RECORD(ACT_WAIT_GRP, spdid, e ? e->extern_id : 0, cos_get_thd_id(), 0);
		if (NULL == g) goto err;

		/* gather multiple events */
		do {
			if (__evt_grp_read_noblock(g, &e)) goto err;
			if (NULL != e) {
				((long*)data->mem)[evt_gathered] = e->extern_id;
				evt_gathered++;
			}
		} while (e && evt_gathered < evt_max);

		/* return them if they were gathered */
		if (evt_gathered > 0) {
			lock_release(&evt_lock);
			return evt_gathered;
		}

		/* 
		 * otherwise sleep till there is an event (first we
		 * need to call evt_grp_read to set the blocked
		 * status)
		 */
		if (__evt_grp_read(g, &e)) goto err;
		assert(NULL == e);
		lock_release(&evt_lock);
		ACT_RECORD(ACT_SLEEP, spdid, 0, cos_get_thd_id(), 0);
		if (0 > sched_block(cos_spd_id(), 0)) BUG();
	}
err:
	lock_release(&evt_lock);
	return -1; 
	
}

td_t 
tsplit(spdid_t spdid, td_t tid, char *param, int len, 
       tor_flags_t tflags, long evtid)
{
	td_t ret = -ENOMEM, ntd;
	struct torrent *t;

	if (tid != td_root) return -EINVAL;
	ntd = parent_tsplit(cos_spd_id(), tid, param, len, tflags, evtid);
	if (ntd <= 0) ERR_THROW(ntd, err);

	t = tor_alloc((void*)ntd, tflags);
	if (!t) ERR_THROW(-ENOMEM, err);
	ret = t->td;
err:
	return ret;
}

static int create_thd(const char *pri)
{
	struct cos_array *data;
	int event_thd;
	int sz = strlen(pri) + 1;
    
	data = cos_argreg_alloc(sizeof(struct cos_array) + sz);
	assert(data);
	strcpy(&data->mem[0], pri);
	//data->sz = 4;
	data->sz = sz;

	if (0 > (event_thd = sched_create_thread(cos_spd_id(), data))) assert(0);
	cos_argreg_free(data);
    
	return event_thd;
}

static void
rd_recover_state(struct rec_data_lk *rd)
{
	assert(rd && rd->c_lkid);
	printc("thd %d is creating a new server side lock id\n", cos_get_thd_id());

	struct rec_data_lk *tmp;
	int tmp_lkid = lock_component_alloc(cos_spd_id());
	assert(tmp_lkid);
	
	printc("thd %d is creating a new server side lock id %d\n", 
	       cos_get_thd_id(), tmp_lkid);
	assert((tmp = rdlk_lookup(tmp_lkid)));
	rd->s_lkid = tmp->s_lkid;
	rdlk_dealloc(tmp_lkid);
	
	return;
}

static inline void
mm_init(void)
{
	printc("core %ld: mm init as thread %d\n", cos_cpuid(), cos_get_thd_id());

	/* Expanding VAS. */
	printc("mm expanding %lu MBs @ %p\n", (NREGIONS-1) * round_up_to_pgd_page(1) / 1024 / 1024, 
	       (void *)round_up_to_pgd_page((unsigned long)&cos_comp_info.cos_poly[1]));
	if (cos_vas_cntl(COS_VAS_SPD_EXPAND, cos_spd_id(), 
			 round_up_to_pgd_page((unsigned long)&cos_comp_info.cos_poly[1]), 
			 (NREGIONS-1) * round_up_to_pgd_page(1))) {
		printc("MM could not expand VAS\n");
		BUG();
	}

	frame_init();
	printc("core %ld: mm init done\n", cos_cpuid());
}