mcache.cc

柯老师网站上找到的

		} else if (strcmp(argv[1], "stream-received") == 0) {
			// We've got the entire page, unlock it
			MediaPage *pg = (MediaPage*)mpool()->get_page(argv[2]);
			assert(pg != NULL);
			pg->unlock();
			// XXX Should we clear all "last" flag of segments??
#ifdef MCACHE_DEBUG
			// Printing out current buffer status of the page
			char *buf;
			for (int i = 0; i < pg->num_layer(); i++) {
				buf = pg->print_layer(i);
				log("E SEGS p %s l %d %s\n", argv[2], i, buf);
				delete []buf;
			}
#endif
			// Show cache free size
			log("E SIZ n %d z %d t %d\n", mpool()->num_pages(),
			    mpool()->usedsize(), mpool()->maxsize());
			return TCL_OK;
		}
	} else if (argc == 5) {
		if (strcmp(argv[1], "register-client") == 0) {
			// <server> register-client <app> <client> <pageid>
			TclObject *a = TclObject::lookup(argv[2]);
			assert(a != NULL);
			int newEntry;
			Tcl_HashEntry *he = Tcl_CreateHashEntry(cmap_, 
					(const char *)a, &newEntry);
			if (he == NULL) {
				tcl.add_errorf("cannot create hash entry");
				return TCL_ERROR;
			}
			if (!newEntry) {
				tcl.add_errorf("duplicate connection");
				return TCL_ERROR;
			}
			RegInfo *p = new RegInfo;
			p->client_ = (HttpApp*)TclObject::lookup(argv[3]);
			assert(p->client_ != NULL);
			strcpy(p->name_, argv[4]);
			Tcl_SetHashValue(he, (ClientData)p);

			// Lock the page while transmitting it to a client
			MediaPage *pg = (MediaPage*)mpool()->get_page(argv[4]);
			assert((pg != NULL) && (pg->type() == MEDIA));
			pg->tlock();

			return TCL_OK;
		} else if (strcmp(argv[1], "unregister-client") == 0) {
			// <server> unregister-client <app> <client> <pageid>
			TclObject *a = TclObject::lookup(argv[2]);
			assert(a != NULL);
			Tcl_HashEntry *he = 
				Tcl_FindHashEntry(cmap_, (const char*)a);
			if (he == NULL) {
				tcl.add_errorf("cannot find hash entry");
				return TCL_ERROR;
			}
			RegInfo *ri = (RegInfo*)Tcl_GetHashValue(he);
			// Update hit count
			mpool()->hc_update(argv[4], ri->hl_);
#ifdef MCACHE_DEBUG
			printf("Cache %d hit counts: \n", id_);
			mpool()->dump_hclist();
#endif
			// Dump per-connection statistics
			for (int i = 0; i <= ri->hl_; i++)
				log("E STAT p %s l %d d %d e %d p %d\n",
				    ri->name_, i, ri->db_[i], ri->eb_[i], 
				    ri->pb_[i]);
			delete ri;
			Tcl_DeleteHashEntry(he);

			// Lock the page while transmitting it to a client
			MediaPage *pg = (MediaPage*)mpool()->get_page(argv[4]);
			assert((pg != NULL) && (pg->type() == MEDIA));
			pg->tunlock();

			return TCL_OK;
		}
	}

	return HttpCache::command(argc, argv);
}



//----------------------------------------------------------------------
// Media web client 
//   Use C++ interface to records quality of received stream.
// NOTE: 
//   It has OTcl inheritance, but no C++ inheritance!
//----------------------------------------------------------------------

static class HttpMediaClientClass : public TclClass {
public:
	HttpMediaClientClass() : TclClass("Http/Client/Media") {}
	TclObject* create(int, const char*const*) {
		return (new MediaClient());
	}
} class_httpmediaclient;

// Records the quality of stream received
void MediaClient::process_data(int size, AppData* data)
{
	assert(data != NULL);

	switch (data->type()) {
	case MEDIA_DATA: {
		HttpMediaData* d = (HttpMediaData*)data;
		// XXX Don't pass any data to page pool!!
   		if (mpool()->add_segment(d->page(), d->layer(), 
   					 MediaSegment(*d)) == -1) {
   			fprintf(stderr, 
  "MediaCache %s gets a segment for an unknown page %s\n", name(), d->page());
//     			abort();
   		}
		// Note: we store the page only to produce some statistics
		// later so that we need not do postprocessing of traces.
#if 1
    		log("C RSEG p %s l %d s %d e %d z %d\n", 
    		    d->page(), d->layer(), d->st(), d->et(), d->datasize());
#endif
		break;
	}
	default:
		HttpClient::process_data(size, data);
	}
}

int MediaClient::command(int argc, const char*const* argv)
{
	if (argc == 3) {
		if (strcmp(argv[1], "stream-received") == 0) {
			// XXX This is the place to do statistics collection
			// about quality of received stream.
			// 
			// Dump delivered quality log
			MediaPage *pg = (MediaPage*)mpool()->get_page(argv[2]);
			assert(pg != NULL);
			// Printing out current buffer status of the page
			char *buf;
			for (int i = 0; i < pg->num_layer(); i++) {
				buf = pg->print_layer(i);
				if (strlen(buf) > 0) 
					log("C SEGS p %s l %d %s\n", 
					    argv[2], i, buf);
				delete []buf;
			}
			// then delete the stream from buffer
			mpool()->force_remove(argv[2]);
			return TCL_OK;
		}
	}
	return HttpClient::command(argc, argv);
}



//----------------------------------------------------------------------
// Multimedia web server
//----------------------------------------------------------------------

static class MediaServerClass : public TclClass {
public:
	MediaServerClass() : TclClass("Http/Server/Media") {}
	TclObject* create(int, const char*const*) {
		return (new MediaServer());
	}
} class_mediaserver;

MediaServer::MediaServer() : HttpServer() 
{
	pref_ = new Tcl_HashTable;
	Tcl_InitHashTable(pref_, 2);
	cmap_ = new Tcl_HashTable;
	Tcl_InitHashTable(cmap_, TCL_ONE_WORD_KEYS);
}

MediaServer::~MediaServer() 
{
	Tcl_HashEntry *he;
	Tcl_HashSearch hs;
	if (pref_ != NULL) {
		for (he = Tcl_FirstHashEntry(pref_, &hs);  he != NULL;
		     he = Tcl_NextHashEntry(&hs)) {
			PrefInfo *pi = (PrefInfo*)Tcl_GetHashValue(he);
			pi->sl_->destroy();
			delete pi->sl_;
		}
		Tcl_DeleteHashTable(pref_);
		delete pref_;
	}
	if (cmap_ != NULL) {
		for (he = Tcl_FirstHashEntry(cmap_, &hs);  he != NULL;
		     he = Tcl_NextHashEntry(&hs))
			delete (RegInfo*)Tcl_GetHashValue(he);
		Tcl_DeleteHashTable(cmap_);
		delete cmap_;
	}
}

// Return the next segment to be sent to a particular application
MediaSegment MediaServer::get_next_segment(MediaRequest *r, Application*& ci)
{
	MediaPage* pg = (MediaPage*)pool_->get_page(r->name());
	assert(pg != NULL);

	// XXX Extremely hacky way to map media app names to 
	// HTTP connections. Should maintain another hash table for this.
	RegInfo *ri = get_reginfo(r->app());
	assert(ri != NULL);
	PrefInfoQ* q = get_piq(r->name(), ri->client_);

	// We are not on the prefetching list, send a normal data segment
	if ((q == NULL) || (q->is_empty())) {
		MediaSegment s1(r->st(), r->et());
		return pg->next_overlap(r->layer(), s1);
	}

	// Cycle through the prefetched segments that we need to send
	int found = 0;
	int searched = 0;
	PrefInfo *pi; 
	while (!found) {
		PrefInfoE *pe = q->dequeue();
		pi = pe->data();
		q->enqueue(pe);
		// If there's a pending segment in any layer, send it
		for (int i = 0; i < pg->num_layer(); i++) 
			if (pi->sl_[i].length() > 0) 
				found = 1;
		// If no pending prefetched segments, return empty
		if (searched++ == q->size()) 
			return MediaSegment(0, 0);
	}

	// Send a segment from the prefetching list. Only use the data size
	// included in the request.
	MediaSegmentList *p = pi->sl_;
	// Set return conid
	ci = pi->conid_;

	// Find one available segment in prefetching list if there is none
	// in the given layer
	int l = r->layer(), i = 0;
	MediaSegment res;
	while ((res.datasize() == 0) && (i < pg->num_layer())) {
		// next() doesn't work. Need a method that returns the 
		// *FIRST* non-empty segment which satisfies the size 
		// constraint.
		res = p[l].get_nextseg(MediaSegment(0, r->datasize()));
		i++;
		l = (l+1) % pg->num_layer();
	}
	// XXX We must do boundary check of the prefetched segments to make
	// sure that the start and end offsets are valid!
	if (res.start() < 0) 
		res.set_start(0);
	if (res.end() > pg->layer_size(l))
		res.set_end(pg->layer_size(l));
	if (res.datasize() > 0) {
		// XXX We may end up getting data from another layer!!
		l = (l-1+pg->num_layer()) % pg->num_layer();
		if (l != r->layer())
			r->set_layer(l);
		// We may not be able to get the specified data size, due 
		// to arbitrary stream lengths
		//assert(res.datasize() == r->datasize());
		p[r->layer()].evict_head(r->datasize());
	}
	// Set the prefetching flag of this segment
	res.set_pref();
	return res;
}

// Similar to MediaCache::get_data(), but ignore segment availability checking
AppData* MediaServer::get_data(int& size, AppData *req)
{
	assert((req != NULL) && (req->type() == MEDIA_REQUEST));
	MediaRequest *r = (MediaRequest *)req;
	Application* conid = NULL;

	if (r->request() == MEDIAREQ_GETSEG) {
		// Get a new data segment
		MediaSegment s2 = get_next_segment(r, conid);
		HttpMediaData *p;
		if (s2.datasize() == 0) {
			// No more data available for this layer, most likely
			// it's because this layer is finished.
			size = 0;
			p = new HttpMediaData(name(), r->name(),
					      r->layer(), 0, 0);
		} else {
			size = s2.datasize();
			p = new HttpMediaData(name(), r->name(),
					   r->layer(), s2.start(), s2.end());
		}
		if (s2.is_last()) {
			p->set_last();
			// Tear down the connection after we've sent the last
			// segment of the base layer and are requested again.
			if ((s2.datasize() == 0) && (r->layer() == 0))
				p->set_finish();
		}
		if (s2.is_pref()) {
			// Add connection id into returned data
			p->set_conid(conid);
			p->set_pref();
		}
		return p;
	} else if (r->request() == MEDIAREQ_CHECKSEG) 
		// We don't need to return anything, so just NULL
		return NULL;
	else {
		fprintf(stderr, 
		       "MediaServer %s gets an unknown MediaRequest type %d\n",
			name(), r->request());
		abort();
	}
	/*NOTREACHED*/
	return NULL; // Make msvc happy
}

int MediaServer::command(int argc, const char*const* argv)
{
	Tcl& tcl = Tcl::instance();
	if (argc == 3) {
		if (strcmp(argv[1], "is-media-page") == 0) {
			ClientPage *pg = pool_->get_page(argv[2]);
			if (pg && (pg->type() == MEDIA))
				tcl.result("1");
			else 
				tcl.result("0");
			return TCL_OK;
		}
	} else if (argc == 5) { 
		if (strcmp(argv[1], "stop-prefetching") == 0) {
			/*
			 * <server> stop-prefetching <Client> <conid> <pagenum>
			 */
			TclObject *a = TclObject::lookup(argv[2]);
			assert(a != NULL);
			int tmp[2];
			tmp[0] = (int)a;
			tmp[1] = atoi(argv[4]);
			Tcl_HashEntry *he = 
				Tcl_FindHashEntry(pref_, (const char*)tmp);
			if (he == NULL) {
				tcl.add_errorf(
				  "Server %d cannot stop prefetching!\n", id_);
				return TCL_ERROR;
			}
			a = TclObject::lookup(argv[3]);
			assert(a != NULL);
			PrefInfoQ *q = (PrefInfoQ*)Tcl_GetHashValue(he);
			PrefInfoE *pe = find_prefinfo(q, (Application*)a);
			assert(pe != NULL);
			PrefInfo *pi = pe->data();
			MediaSegmentList *p = pi->sl_;
			assert(p != NULL);
			for (int i = 0; i < MAX_LAYER; i++)
				p[i].destroy();
			delete []p;
			delete pi;
			q->detach(pe);
			delete pe;
			// If no more prefetching streams left for this client,
			// delete all the information.
			// Return 0 means that we still have prefetching 
			// clients left, don't tear down the channel yet. 
			// Otherwise return 1. 
			int res = 0;
			if (q->is_empty()) {
				delete q;
				Tcl_DeleteHashEntry(he);
				res = 1;
			}
			tcl.resultf("%d", res);
			return (TCL_OK);
		} else if (strcmp(argv[1], "register-client") == 0) {
			// <cache> register-client <app> <client> <pageid>
			TclObject *a = TclObject::lookup(argv[2]);
			assert(a != NULL);
			int newEntry;
			Tcl_HashEntry *he = Tcl_CreateHashEntry(cmap_, 
					(const char *)a, &newEntry);
			if (he == NULL) {
				tcl.add_errorf("cannot create hash entry");
				return TCL_ERROR;
			}
			if (!newEntry) {
				tcl.add_errorf("duplicate connection");
				return TCL_ERROR;
			}
			RegInfo *p = new RegInfo;
			p->client_ = (HttpApp*)TclObject::lookup(argv[3]);
			assert(p->client_ != NULL);
			strcpy(p->name_, argv[4]);
			Tcl_SetHashValue(he, (ClientData)p);
			return TCL_OK;
		} else if (strcmp(argv[1], "unregister-client") == 0) {
			// <cache> unregister-client <app> <client> <pageid>
			TclObject *a = TclObject::lookup(argv[2]);
			assert(a != NULL);
			Tcl_HashEntry *he = 
				Tcl_FindHashEntry(cmap_, (const char*)a);
			if (he == NULL) {
				tcl.add_errorf("cannot find hash entry");
				return TCL_ERROR;
			}
			RegInfo *p = (RegInfo*)Tcl_GetHashValue(he);
			delete p;
			Tcl_DeleteHashEntry(he);
			return TCL_OK;
		}
	} else {
		if (strcmp(argv[1], "enter-page") == 0) {
			ClientPage *pg = pool_->enter_page(argc, argv);
			if (pg == NULL)
				return TCL_ERROR;
			if (pg->type() == MEDIA) 
				((MediaPage*)pg)->create();
			// Unlock the page after creation
			((MediaPage*)pg)->unlock(); 
			return TCL_OK;
		} else if (strcmp(argv[1], "register-prefetch") == 0) {
			/*
			 * <server> register-prefetch <client> <pagenum> 
			 * 	<conid> <layer> {<segments>}
			 * Registers a list of segments to be prefetched by 
			 * <client>, where each <segment> is a pair of 
			 * (start, end). <pagenum> should be pageid without 
			 * preceding [server:] prefix.
			 * 
			 * <conid> is the OTcl name of the original client 
			 * who requested the page. This is used for the cache
			 * to get statistics about a particular connection.
			 * 
			 * <client> is the requestor of the stream.
			 */
			TclObject *a = TclObject::lookup(argv[2]);
			assert(a != NULL);
			int newEntry = 1;
			int tmp[2];
			tmp[0] = (int)a;
			tmp[1] = atoi(argv[3]);
			// Map <cache_ptr><conid> to a pref entry
			Tcl_HashEntry *he = Tcl_CreateHashEntry(pref_, 
					(const char*)tmp, &newEntry);
			if (he == NULL) {
				fprintf(stderr, "Cannot create entry.\n");
				return TCL_ERROR;
			}
			PrefInfo *pi;
			PrefInfoE *pe;
			PrefInfoQ *q; 
			MediaSegmentList *p;
			a = TclObject::lookup(argv[4]);
			if (newEntry) {
				q = new PrefInfoQ;
				Tcl_SetHashValue(he, (ClientData)q);
				pe = NULL;
			} else {
				q = (PrefInfoQ *)Tcl_GetHashValue(he);
				pe = find_prefinfo(q, (Application*)a);
			}
			if (pe == NULL) {
				pi = new PrefInfo;
				pi->conid_ = (Application*)a;
				p = pi->sl_ = new MediaSegmentList[MAX_LAYER];
				q->enqueue(new PrefInfoE(pi));
			} else {
				pi = pe->data();
				p = pi->sl_;
			}
			assert((pi != NULL) && (p != NULL));
			// Preempt all old requests because they 
			// cannot reach the cache "in time"
			int layer = atoi(argv[5]);
			p[layer].destroy();
			// Add segments into prefetching list
			assert(argc % 2 == 0);
			for (int i = 6; i < argc; i+=2)
				p[layer].add(MediaSegment(atoi(argv[i]), 
							  atoi(argv[i+1])));
			return TCL_OK;
		}
	}
			
	return HttpServer::command(argc, argv);
}