#include "Cache.h"
#include <iostream.h>
#include <except.h>
#define BUFFERING 100

class mem_counter
{
int size;
int fr;
public:
    mem_counter():size(0),fr(0){}
    void add(int z){size+=z;}
    void free(int z){fr+=z;}
    ~mem_counter()
    {
//	cout<<"Allocated "<<size/1024<<" Kbytes"<<endl;
//	cout<<"Freed "<<fr/1024<<" Kbytes"<<endl;
    }
};

static mem_counter mc;
    
struct thread_arg
{
    int fd;
    req* req_buf;
    int* buffers_busy;
    int* quit;
    pthread_mutex_t* mutex_in;
    pthread_cond_t* cond_in;
    pthread_mutex_t* mutex_out;
    pthread_cond_t* cond_out;
};    
/************************************************************

 This function looks through array of req's and searches
 for waiting req with highest priority ( lowest number of 
 complete requests before this in the same stream ).
 
*************************************************************/
static int minimal(req* req_buf)
{
    int ans=0;
    int streams=-1;
    int i;
// Determine number of streams
#warning FIXME
    for(i=0; i<BUFFERING; i++)
	if(req_buf[i].st!=req::BUFFER_CLEAN)
	    if((signed)req_buf[i].id>streams)
		streams=req_buf[i].id;
    if(streams<0)
    {
	return 0;//req_buf is completely clean
    }
// Init	priority table
    int* bases=new int[streams+1];
    mc.add((streams+1)*4);
    for(i=0; i<streams; i++)bases[i]=-1;
    for(i=0; i<BUFFERING; i++)
	if(req_buf[i].st!=req::BUFFER_CLEAN)
	{
	    if(req_buf[i].id<0)cerr<<"Oops: id "<<req_buf[i].id<<endl;
	    if(req_buf[i].id>streams)cerr<<"Oops: id "<<req_buf[i].id<<endl;
	    if((req_buf[i].position<bases[req_buf[i].id]) || (bases[req_buf[i].id]==-1))
    		bases[req_buf[i].id]=req_buf[i].position;
	}	
// Find lowest waiting chunk
    int val=-1;
    for(i=0; i<BUFFERING; i++)
	if(req_buf[i].st==req::BUFFER_WAITING)
	    if((req_buf[i].position-bases[req_buf[i].id]<val) || (val==-1))
	    {
		ans=i;
		val=req_buf[i].position-bases[req_buf[i].id];
	    }
    delete bases;
    mc.free((streams+1)*4);
//    cerr<<"Waiting req_buf "<<ans<<endl;
    return ans;	    
}
void* readfunc(void* arg)
{
    thread_arg a;
    memcpy(&a, arg, sizeof(a));
    delete (thread_arg*)arg;
    int qz=0;
    while(1)
    {
        if(*(a.quit))return 0;
        
	pthread_mutex_lock(a.mutex_out);
	if(*(a.buffers_busy)==0)
	    pthread_cond_wait(a.cond_in, a.mutex_out);
	pthread_mutex_unlock(a.mutex_out);
	
	int i=minimal(a.req_buf);
	pthread_mutex_lock(a.mutex_out);
	if(a.req_buf[i].st!=req::BUFFER_WAITING)
	{
	    pthread_mutex_unlock(a.mutex_out);
	    continue;
    	}
    	a.req_buf[i].st=req::BUFFER_WRITING;
	if(a.req_buf[i].memory)
	{
	    printf("Logic error: buffer with nonzero memory\n");
	}    
	a.req_buf[i].memory=new char[a.req_buf[i].size & 0x7FFFFFFF];
	mc.add(a.req_buf[i].size & 0x7FFFFFFF);
	pthread_mutex_unlock(a.mutex_out);
	lseek64(a.fd, a.req_buf[i].offset & 0xFFFFFFFFFFFFLL, SEEK_SET);
	read(a.fd, a.req_buf[i].memory, a.req_buf[i].size & 0x7FFFFFFF);
  	pthread_mutex_lock(a.mutex_out);
	a.req_buf[i].st=req::BUFFER_READY;
	(*(a.buffers_busy))--;
	pthread_mutex_unlock(a.mutex_out);
	pthread_cond_broadcast(a.cond_out);
    }		    
}
Cache::Cache(Unsigned size): m_size(size), m_used(0), m_quit(0), buffers_busy(0), m_status(CLEAR),
    cache_access(0), cache_right(0), cache_miss(0)
{
    memset(m_tables, 0, sizeof(m_tables));
    req_buf=new req[size];
    
}

InputStream::InputStream(int fd, int fd_async):m_fd(fd),m_fd_async(fd_async),cache(new Cache(BUFFERING))
{
    int pos=lseek64(m_fd, 0, SEEK_CUR);
    length=lseek64(m_fd, 0, SEEK_END);
    seek(pos);
}
int Cache::AddStream(Unsigned id, AVIIndexEntry2 *table, Unsigned position, Unsigned length)
{
    printf("Cache: Adding stream %d, table %X ( %d entries )\n", (int)id, (int)table, (int)length);
    if(id>=MAXSTREAMS)
    {
        printf("Only %d cached streams are supported\n", MAXSTREAMS);
	return -1;
    }
    if(position>=length)
	return -1;
    m_tables [id]=table;
    m_lengths [id]=length;
    m_positions [id]=position;
    return 0;
}

int Cache::Create(int m_fd)
{
    printf("Cache:   Creating cache for file descriptor %d\n", m_fd);
    thread_arg* str;
    int result;
    int i;
    pthread_mutex_init(&mutex_in, 0);
    pthread_mutex_init(&mutex_out, 0);
    for(i=0; i<BUFFERING; i++)
    {
	req_buf[i].st=req::BUFFER_CLEAN;
	req_buf[i].memory=0;
    }	
    result=pthread_cond_init(&cond_in, NULL);
    if(result!=0)perror("Creating cond_in");
    result=pthread_cond_init(&cond_out, NULL);
    if(result!=0)perror("Creating cond_out");
    str=new thread_arg;
    str->fd=m_fd;
    str->mutex_in=&mutex_in;
    str->mutex_out=&mutex_out;
    str->cond_in=&cond_in;
    str->cond_out=&cond_out;
    str->req_buf=req_buf;
    str->buffers_busy=&buffers_busy;
    str->quit=&m_quit;
    result=pthread_create(&thread, 0, &readfunc, str);
    if(result!=0)perror("Creating thread");
    m_status=ASYNC;
    return 0;
//    Sync();
//    usleep(100000);
}
Cache::~Cache()
{
    m_quit=1;
    if(m_status==ASYNC)
    {
	pthread_cond_broadcast(&cond_in);
	pthread_join(thread, NULL);
	pthread_cond_destroy(&cond_in);	    
	pthread_cond_destroy(&cond_out);	    
	pthread_mutex_destroy(&mutex_in);	    
        pthread_mutex_destroy(&mutex_out);	    
	for(int i=0; i<m_size; i++)
	    if(req_buf[i].memory)
	    {
		delete req_buf[i].memory;
		mc.free(req_buf[i].size & 0x7FFFFFFF);
	    }	
	delete req_buf;	    
    }
    if(cache_access==0)return;
    printf("Destroying cache. Total accesses %d, hits %.2f%%, misses %.2f%%, errors %.2f%%\n",
	cache_access, 100.*double(cache_right-cache_miss)/cache_access, 100.*double(cache_miss)/cache_access,
	    100.*double(cache_access-cache_right)/cache_access);
}
InputStream::~InputStream()
{
    if(cache)
	delete cache;
    cache=0;
}
//#warning
int Cache::Read(char* buffer, Unsigned id, Unsigned position, Unsigned size, Unsigned in_pos)
// offset_t offset, char* buffer, Unsigned size)
{
    int t1, t2;
//    t1=lcount();
//    printf("Cache: read(buffer %x, id %d, pos %d, size %d, in_pos %d)\n",
//	buffer, id, position, size, in_pos);
    cache_access++;
    if(id>=MAXSTREAMS)return E_ERROR;
    if(position>=m_lengths[id])return E_ERROR;
    cache_right++;
    __int64 offset=m_tables[id][position].pos;
    Unsigned max_size=(m_tables[id][position].size & 0x7FFFFFFF) +8;
//    printf("Actual size %d\n", max_size-8);
//    printf("Offset %Lx, size %Lx", offset, max_size);
    m_positions[id]=position;
    Update();
    int freebuf=0;
    int waiting=0;
    int ready=0;
in1:
    for(int i=0; i<m_size; i++)
    {
	switch(req_buf[i].st)
	{
	    case req::BUFFER_CLEAN:
		freebuf++;
		break;
	    case req::BUFFER_WAITING:
		waiting++;
		break;
	    case req::BUFFER_READY:
		ready++;
		break;
	}
	if(req_buf[i].offset==offset && req_buf[i].size==max_size && req_buf[i].st!=req::BUFFER_CLEAN)
	{
//	    printf("Waiting for cond_out at %X, buffer %d, "
//	    "offset %d, size %d, status %d\n", lcount(), i,
//	    req_buf[i].offset, req_buf[i].size, req_buf[i].st);
	    if(req_buf[i].st!=req::BUFFER_READY)
	    {
//		printf("Locking mutex_out\n");
	        pthread_mutex_lock(&mutex_out);
//		printf("Locked mutex_out\n");
		while(req_buf[i].st!=req::BUFFER_READY)
		    pthread_cond_wait(&cond_out, &mutex_out);
//		printf("Unlocking mutex_out\n");
	    	pthread_mutex_unlock(&mutex_out);
//		printf("Unlocked mutex_out\n");
//		printf("Received cond_out at %X\n", lcount());
//		printf("Status is %d\n", req_buf[i].st);
	    }
//	    printf("Starting read\n");
	    int chunk_id=*(int*)req_buf[i].memory;
	    if(StreamFromFOURCC(chunk_id)!=id)
	    {
		printf("WARNING: Read(): FOURCC mismatch ( received %X )\n", chunk_id);
	    }	
	    if(size>max_size-in_pos-8)
		size=max_size-in_pos-8;
	    memcpy(buffer, req_buf[i].memory+in_pos+8, size);
//	    printf("Ending read\n");
//	    printf("Deleting\n");
//	    pthread_mutex_lock(&mutex_out);
//	    delete req_buf[i].memory;
//	    req_buf[i].memory=0;
//	    req_buf[i].st=req::BUFFER_CLEAN;
//	    t2=lcount();
//	    pthread_mutex_unlock(&mutex_out);
//	    printf("Success\n");
//	    printf("Read: read %d bytes at %f ms\n", 
//	     req_buf[i].size, (t2-t1)/550000.);
	    return size;
	}    	
    }
//    printf("Read: no buffer found ( %d free, %d waiting, %d ready )\n", freebuf, waiting, ready);
    Prefetch(id, position);
    cache_miss++;
    goto in1;
}

int InputStream::Read(char* buffer, Unsigned size)
{
    int ans=read(m_fd, buffer, size);
    return ans;
}

    
int Cache::Prefetch(Unsigned id, Unsigned position)
//offset_t offset, Unsigned size)
{
//    printf("Prefetch(%d, %d)\n", id, position);
    if(id>=MAXSTREAMS)return E_ERROR;
    if(position>=m_lengths[id])return E_ERROR;
    __int64 offset=m_tables[id][position].pos;
    Unsigned size=(m_tables[id][position].size & 0x7FFFFFFF) +8;
//    Unsigned size=m_tables[id][position].size+8;

/*    offset_t offset=m_tables[id][position].dwChunkOffset;
    Unsigned size=m_tables[id][position].dwChunkLength+8;*/
//    printf("Prefetch():resolved to %d,%d\n", offset, size);
    int i;
    for(i=0; i<m_size; i++)
    {
	if(req_buf[i].st==req::BUFFER_CLEAN)
	{
//	    printf("Locking mutex_in\n");
	    pthread_mutex_lock(&mutex_out);
	    req_buf[i].offset=offset;
	    req_buf[i].size=size;
	    req_buf[i].id=id;
	    req_buf[i].position=position;
	    req_buf[i].st=req::BUFFER_WAITING;
//	    printf("Broadcasting cond_in at %X\n", lcount());
	    buffers_busy++;
	    pthread_cond_broadcast(&cond_in);
	    pthread_mutex_unlock(&mutex_out);
//	    printf("Unlocked mutex_in\n");
	    sched_yield();
	    return 0;
	}
    }	    
    Clear();
    Prefetch(id,position);
    /*
    printf("Prefetch: can't find free buffer\n");
    printf("Dumping buffers: \n");
    for(i=0; i<m_size; i++)printf("%.2d: offset 0x%Lx, size %d, id %d, position %d, status %s\n",
	    i, req_buf[i].offset, req_buf[i].size & 0x7fffffff, req_buf[i].id, req_buf[i].position,
		    (req_buf[i].st==req::BUFFER_WAITING?"WAITING":
		    (req_buf[i].st==req::BUFFER_WRITING?"WRITING":
		    (req_buf[i].st==req::BUFFER_CLEAN?"CLEAN":
		    (req_buf[i].st==req::BUFFER_READY?"READY":
		    "UNKNOWN")))));
    printf("Requested buffer for %d/%d\n", id, position);		    
    exit(1);
    return E_ERROR;
    */
} 


int Cache::Clear()
{
    int freebuf=0, waiting=0, ready=0;
//    printf("Clearing cache\n");
    pthread_mutex_lock(&mutex_out);
    for(int i=0; i<m_size; i++)
    {
	if(req_buf[i].st==req::BUFFER_WRITING)
	    pthread_cond_wait(&cond_out, &mutex_out);
	    else
	if(req_buf[i].st==req::BUFFER_WAITING)
	    req_buf[i].st=req::BUFFER_CLEAN;
	if(req_buf[i].st==req::BUFFER_READY)
	{
	    delete req_buf[i].memory;
	    mc.free(req_buf[i].size & 0x7FFFFFFF);
	    req_buf[i].memory=0;
    	    req_buf[i].st=req::BUFFER_CLEAN;
	}	    
    }
//    pthread_mutex_lock(&mutex_in);
    buffers_busy=0;
//    pthread_mutex_unlock(&mutex_in);
    int i;
    for(i=0; i<m_size; i++)
    {
	switch(req_buf[i].st)
	{
	    case req::BUFFER_CLEAN:
		freebuf++;
		break;
	    case req::BUFFER_WAITING:
		waiting++;
		break;
	    case req::BUFFER_READY:
		ready++;
		break;
	}
    }	
    for(i=0; i<MAXSTREAMS; i++)
	m_positions[i]=0;    
    pthread_mutex_unlock(&mutex_out);
//    printf("Clearing cache OK: %d clean, %d waiting, %d ready\n", freebuf, waiting, ready);
    Update();
    usleep(100000);
//    Sync();
    return 0;    
}

int Cache::Update()
{
    pthread_mutex_lock(&mutex_out);
    int free_reqs=0;
    char** flags;
    int i;
    static int caller=0;
    flags=new char*[MAXSTREAMS];
    mc.add(MAXSTREAMS*4);
    for(i=0; i<MAXSTREAMS; i++)
    {
	flags[i]=new char[m_size];
	mc.add(m_size);
	memset(flags[i], 0, m_size);
    }    
    for(i=0; i<m_size; i++)
    {
	if(req_buf[i].st==req::BUFFER_READY)
	if((unsigned)req_buf[i].position<m_positions[req_buf[i].id])
	{
//	    printf("Deleting buffer %d\n", i);
	    delete req_buf[i].memory;
	    mc.free(req_buf[i].size & 0x7FFFFFFF);
//	    printf("Deleting buffer %d OK\n", i);
	    req_buf[i].memory=0;
    	    req_buf[i].st=req::BUFFER_CLEAN;
	}
	if(req_buf[i].st==req::BUFFER_CLEAN)
	    free_reqs++;
	else
	{
	    int th=req_buf[i].position-m_positions[req_buf[i].id];
	    if(th<m_size)
		flags[req_buf[i].id][th]=1;
	}	
    }
    caller++;
    if(caller%100==0)
    {
	int waiting=0, writing=0, ready=0;
	int priority=-1;
	int cpr;
	for(i=0; i<m_size; i++)
	{
	    if(req_buf[i].st==req::BUFFER_CLEAN)continue;
//	    printf("%d: stream %d, pos %d, ", i, req_buf[i].id, req_buf[i].position);
	    switch(req_buf[i].st)
	    {
		case req::BUFFER_WAITING:
		    waiting++;
		    cpr=req_buf[i].position-m_positions[req_buf[i].id];
		    if(priority<0)priority=cpr;
		    if(priority>cpr)priority=cpr;
		    break;
//		    printf("waiting\n");break;
		case req::BUFFER_WRITING:
		    writing++;
		    break;
//		    printf("writing\n");break;
		case req::BUFFER_READY:
		    ready++;
		    break;
//		    printf("ready\n");break;
	    }
	}
//	if(waiting!=0)
//	    printf("Cache status: %d waiting ( lowest priority %d ), %d writing, %d ready, %d clean\n", 
//		waiting, priority, writing, ready, free_reqs);
    }	    	    
    pthread_mutex_unlock(&mutex_out);
    int c_id=0, c_pos=1;
//    printf("%d free reqs\n", free_reqs);
    free_reqs-=2;
    while(free_reqs>0)
    {
	if(m_tables[c_id]!=0)
	{
    	    if(c_pos+m_positions[c_id]>=m_lengths[c_id])
		break;
	    if(c_pos>=m_size)
		break;
	    if(flags[c_id][c_pos]==0)
    	    {
		if(m_lengths[c_id]<=(m_positions[c_id]+c_pos))
		    continue;
		Prefetch(c_id, m_positions[c_id]+c_pos);
		free_reqs--;	
	    }
	}    
	c_id++;
	if(c_id==MAXSTREAMS)
	{	
	    c_id=0;
	    c_pos++;
	}        
    }
    for(i=0; i<MAXSTREAMS; i++)
    {
	delete flags[i];
	mc.free(m_size);
    }
    delete flags;
    mc.free(MAXSTREAMS*4);
    return 0;
}

offset_t InputStream::seek(__int64 offset)
{
    return lseek64(m_fd, offset, SEEK_SET);
}
offset_t InputStream::seek_cur(__int64 offset)
{
    return lseek64(m_fd, offset, SEEK_CUR);
}
offset_t InputStream::pos()
{
    return lseek64(m_fd, 0, SEEK_CUR);
}