void GenericMediaServer::closeAllClientSessionsForServerMediaSession(ServerMediaSession* serverMediaSession) {
    if (serverMediaSession == NULL) return;

    HashTable::Iterator* iter = HashTable::Iterator::create(*fClientSessions);
    GenericMediaServer::ClientSession* clientSession;
    char const* key; // dummy
    while ((clientSession = (GenericMediaServer::ClientSession*)(iter->next(key))) != NULL) {
        if (clientSession->fOurServerMediaSession == serverMediaSession) {
            delete clientSession;
        }
    }
    delete iter;
}

WriteTable* StoreCopy::simdprobeCursor(PageCursor* t, int threadid, WriteTable* ret)
{
	if (ret == NULL) {
		ret = new WriteTable();
		ret->init(sout, outputsize);
	}

	char tmp[sout->getTupleSize()];
	void* tup1;
	void* tup2;
	Page* b2;
	unsigned int curbuc, i;

	HashTable::Iterator it = hashtable.createIterator();

	while (b2 = (atomic ? t->atomicReadNext() : t->readNext())) {

		i = 0;
		while (tup2 = b2->getTupleOffset(i++)) {
			curbuc = _hashfn->hash(s2->asLong(tup2, ja2));
			hashtable.placeIterator(it, curbuc);

			while (tup1 = it.readnext()) {
				if (sbuild->asLong(tup1,0) != s2->asLong(tup2,ja2) ) {
					continue;
				}

#if defined(OUTPUT_ASSEMBLE)
				// copy payload of first tuple to destination
				if (s1->getTupleSize())
					s1->copyTuple(tmp, sbuild->calcOffset(tup1,1));

				// copy each column to destination
				for (unsigned int j=0; j<sel2.size(); ++j)
					sout->writeData(tmp,		// dest
							s1->columns()+j,	// col in output
							s2->calcOffset(tup2, sel2[j]));	// src for this col
#if defined(OUTPUT_WRITE_NORMAL)
				ret->append(tmp);
#endif
#endif

			}
		}
	}
	return ret;
}

void MatroskaDemux::handleEndOfFile() {
  // Iterate through all of our 'demuxed tracks', handling 'end of input' on each one.
  // Hack: Because this can cause the hash table to get modified underneath us, we don't call the handlers until after we've
  // first iterated through all of the tracks.
  unsigned numTracks = fDemuxedTracksTable->numEntries();
  if (numTracks == 0) return;
  MatroskaDemuxedTrack** tracks = new MatroskaDemuxedTrack*[numTracks];

  HashTable::Iterator* iter = HashTable::Iterator::create(*fDemuxedTracksTable);
  unsigned i;
  char const* trackNumber;

  for (i = 0; i < numTracks; ++i) {
    tracks[i] = (MatroskaDemuxedTrack*)iter->next(trackNumber);
  }
  delete iter;

  for (i = 0; i < numTracks; ++i) {
    if (tracks[i] == NULL) continue; // sanity check; shouldn't happen
    FramedSource::handleClosure(tracks[i]);
  }

  delete[] tracks;
}

WriteTable* StoreCopy::realprobeCursor(PageCursor* t, int threadid, WriteTable* ret)
{
	if (ret == NULL) {
		ret = new WriteTable();
		ret->init(sout, outputsize);
	}

	char tmp[sout->getTupleSize()];
	void* tup1;
	void* tup2;
	Page* b2;
	unsigned int curbuc, i;

	HashTable::Iterator it = hashtable.createIterator();

    t->reset(); // added by yipeng to produce output in 1 thread

	while (b2 = (atomic ? t->atomicReadNext() : t->readNext())) {
#ifdef VERBOSE
		cout << "Working on page " << b2 << endl;
                cout << "Thread " << threadid << endl;
#endif
		i = 0;
		while (tup2 = b2->getTupleOffset(i++)) {
#ifdef VERBOSE
			cout << "Joining tuple " << b2 << ":"
				<< setfill('0') << setw(6) << i
				<< " having key " << s2->asLong(tup2, ja2) << endl;
#endif
			curbuc = _hashfn->hash(s2->asLong(tup2, ja2));
#ifdef VERBOSE
			cout << "\twith bucket " << setfill('0') << setw(6) << curbuc << endl;
#endif
			hashtable.placeIterator(it, curbuc);

#ifdef PREFETCH
#warning Only works for 16-byte tuples!
			hashtable.prefetch(_hashfn->hash(
						*(unsigned long long*)(((char*)tup2)+32)
						));
			hashtable.prefetch(_hashfn->hash(
						*(unsigned long long*)(((char*)tup2)+64)
						));
#endif

			while (tup1 = it.readnext()) {
				if (sbuild->asLong(tup1,0) != s2->asLong(tup2,ja2) ) {
					continue;
				}

				// copy payload of first tuple to destination
				if (s1->getTupleSize()) 
					s1->copyTuple(tmp, sbuild->calcOffset(tup1,1));

				// copy each column to destination
				for (unsigned int j=0; j<sel2.size(); ++j)
					sout->writeData(tmp,		// dest
							s1->columns()+j,	// col in output
							s2->calcOffset(tup2, sel2[j]));	// src for this col

				ret->append(tmp);
#ifdef OUTPUT_WRITE_NT
				ret->nontemporalappend16(tmp);
#endif


#if defined(OUTPUT_AGGREGATE)
				aggregator[ (threadid * AGGLEN) +
					+ (sbuild->asLong(tup1,0) & (AGGLEN-1)) ]++;
#endif

#if !defined(OUTPUT_AGGREGATE) && !defined(OUTPUT_ASSEMBLE)
				__asm__ __volatile__ ("nop");
#endif

			}
		}
	}
	return ret;
}