fifo.h

linux下实现视频播放的播放器

/*
 *  Copyright (C) 2005-2007  gulikoza
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

/* $Id$ */

#ifndef FIFO_H
#define FIFO_H

#define MODULE "FIFO"

class FIFO {
protected:
    MUTEX * fifo_lock;
    SDL_semaphore * sem;
    SDL_semaphore * ack;

    bool init;
    struct PACKET {
	unsigned int offset;
	unsigned int size;
	unsigned int pts;
    } *fifo;

    unsigned int read, write;
    unsigned int size;

public:

    FIFO(unsigned int s):init(true),read(s-1),write(s-1),size(s) {
	fifo_lock = MUTEX_CREATE();
	if(fifo_lock == NULL) {
	    fprintf(stderr, "Throwing expception...\n");
	    throw "FIFO: Failed to create mutex";
	}

	sem = SDL_CreateSemaphore(0);
	if(sem == NULL) {
	    MUTEX_DESTROY(fifo_lock);
	    throw "FIFO: Failed to create semaphore sem";
	}

	ack = SDL_CreateSemaphore(0);
	if(ack == NULL) {
	    MUTEX_DESTROY(fifo_lock);
	    SDL_DestroySemaphore(sem);
	    throw "FIFO: Failed to create semaphore ack";
	}

	fifo = (PACKET*)malloc(sizeof(PACKET)*size);
	if(fifo == NULL) {
	    MUTEX_DESTROY(fifo_lock);
	    SDL_DestroySemaphore(sem);
	    SDL_DestroySemaphore(ack);
	    throw "FIFO: Failed to allocate memory";
	}

	fifo[(s-1)].offset = 0;
    }

    bool AddPacket(unsigned int offset, unsigned int pts) {
	MUTEX_LOCK(fifo_lock);

	++write;
	unsigned int inx = write%size;

	while((write > read) && (inx == read%size)) {
	    MUTEX_UNLOCK(fifo_lock);
	    LOG_MSG("Error queue full: %d %d (%d), data dropped!", read, write, inx);
	    return false;
	}

	if(offset >= fifo[(write-1)%size].offset)
	    fifo[(write-1)%size].size = offset-fifo[(write-1)%size].offset;
	fifo[inx].offset = offset;
	fifo[inx].pts = pts;
	fifo[inx].size = 0;
	LOG_MSG("Added packet (%d): %d, pts: %d (previous size: %d)", inx, fifo[inx].offset, fifo[inx].pts, fifo[(write-1)%size].size);
	MUTEX_UNLOCK(fifo_lock);

	SDL_SemPost(sem);

	return true;
    }

    unsigned int GetPacket(unsigned int *data, unsigned int *pts) {

	unsigned int inx, ret;
	if(!init) {
	    ERROR_MSG("FIFO Queue not initialized!");
	    return 0;
	}

#ifdef SEM_SILENT
	SDL_SemWait(sem);
#else
        while(SDL_SemWaitTimeout(sem, 1000) == SDL_MUTEX_TIMEDOUT) {
            ERROR_MSG("Semaphore timeout expired, line: %d, thread: %u", __LINE__, SDL_ThreadID());
	}
#endif
	MUTEX_LOCK(fifo_lock);

	inx = read%size;
	++read;

	ret = fifo[inx].size;
	*pts = fifo[inx].pts;
	*data = fifo[inx].offset;

	LOG_MSG("Returning packet (%d): %d, size: %d, pts: %d", inx, fifo[inx].offset, fifo[inx].size, fifo[inx].pts);
	MUTEX_UNLOCK(fifo_lock);

	if(!SDL_SemValue(ack))
	    SDL_SemPost(ack);
	return ret;
    }

    unsigned int GetCurrentPacket() {

	unsigned int ret;
	MUTEX_LOCK(fifo_lock);
	ret = fifo[write%size].offset;
	MUTEX_UNLOCK(fifo_lock);
	return ret;
    }

    void ResetCurrentPacket(unsigned int offset) {
	// if offset, set current packet size since offset will be set to 0 after
	// if !offset, set current offset to 0

	MUTEX_LOCK(fifo_lock);
	unsigned int inx = write%size;
	if(offset) {
	    fifo[inx].size = offset-fifo[inx].offset;
	} else {
	    fifo[inx].offset = 0;
	}
	MUTEX_UNLOCK(fifo_lock);
    }

    void End(void) {
	init = false;
	SDL_SemPost(sem);
    }

    ~FIFO() {
	MUTEX_DESTROY(fifo_lock);
	SDL_DestroySemaphore(sem);
	SDL_DestroySemaphore(ack);
	free(fifo);
    }
};

#undef MODULE

#define MODULE "QUEUE"

class QUEUE {
protected:
    MUTEX * queue_lock;

    struct LIST {
	char id;
	bool in_use;
	LIST * next;
	LIST():id(-1),in_use(false),next(NULL) { }
    } empty;

    struct {
	LIST * first;
	LIST * last;
    } free, used, ready;

    LIST * available;

    void ListQueues() {
	LIST * pt = free.first;

	while(pt) {
	    LOG_MSG("free id: %d, in_use: %s", pt->id, pt->in_use ? "true" : "false");
	    pt = pt->next;
	}

	pt = used.first;
	while(pt) {
	    LOG_MSG("used id: %d, in_use: %s", pt->id, pt->in_use ? "true" : "false");
	    pt = pt->next;
	}

	pt = ready.first;
	while(pt) {
	    LOG_MSG("ready id: %d, in_use: %s", pt->id, pt->in_use ? "true" : "false");
	    pt = pt->next;
	}

	LOG_MSG("available id: %d", (available) ? available->id : -1);
    }

public:

    QUEUE():available(NULL) {
	queue_lock = MUTEX_CREATE();
	if(queue_lock == NULL) {
	    fprintf(stderr, "Throwing expception...\n");
	    throw "QUEUE: Failed to create mutex";
	}

	used.first = used.last = NULL;
	free.first = free.last = NULL;
	ready.first = ready.last = NULL;
    }

    bool AddFree(char id) {
	MUTEX_LOCK(queue_lock);

	if(id < 0) {
	    if((available == NULL) || (available->id != ~id)) {
		MUTEX_UNLOCK(queue_lock);
		return false;
	    }

	    LOG_MSG("Freeing object %d, available set to %d", available->id,
		    (available->next) ? available->next->id : -1);
	    available = available->next;
#ifdef DEBUG
	    ListQueues();
#endif
	    MUTEX_UNLOCK(queue_lock);
	    return true;
	}

	if((available) && (available->id == -1)) {
	    MUTEX_UNLOCK(queue_lock);
	    available = NULL;
	    return false;
	}

	DEBUG_MSG("Adding %d to the free queue", id);

	if(free.first == NULL) {
	    free.first = new LIST;
	    if(free.first == NULL) {
		ERROR_MSG("Unable to allocate memory!");
		MUTEX_UNLOCK(queue_lock);
		return false;
	    }
	    free.first->id = id;
	    free.last = free.first;
	    MUTEX_UNLOCK(queue_lock);
	    return true;
	}

	free.last->next = new LIST;
	if(free.last->next == NULL) {
	    ERROR_MSG("Unable to allocate memory!");
	    MUTEX_UNLOCK(queue_lock);
	    return false;
	}
	free.last = free.last->next;
	free.last->id = id;
	MUTEX_UNLOCK(queue_lock);
	return true;
    }

    int GetFree() {
	MUTEX_LOCK(queue_lock);
	LIST * pt = ready.first, * p = NULL;
	// Garbage collector, check ready queue for a free entry
	while((pt) && (pt != available)) {
	    if(pt->in_use == false) {
		if(p) p->next = pt->next;
		else ready.first = pt->next;
		pt->next = free.first;
		free.first = pt;
		pt = (p) ? p->next : ready.first;
		LOG_MSG("GBC found %d, moving to free queue", free.first->id);
	    } else {
		p = pt;
		pt = pt->next;
	    }
	}
	if(free.first == NULL) {
	    LOG_MSG("GetFree: free queue empty!");
#ifdef DEBUG
	    ListQueues();
#endif
	    MUTEX_UNLOCK(queue_lock);
	    return -1;
	}

	int id = free.first->id;
	LOG_MSG("Returning %d from the free queue", id);

	if(used.first == NULL) {
	    used.first = used.last = free.first;
	} else {
	    used.last->next = free.first;
	    used.last = free.first;
	}
	free.first = free.first->next;
	used.last->next = NULL;
	MUTEX_UNLOCK(queue_lock);
	return id;
    }

    bool ReturnUsed(char id, bool changed) {
	MUTEX_LOCK(queue_lock);
	LIST * pt, * p = NULL;
	pt = used.first;
	while(pt) {
	    if(pt->id == id) {

		if(p) p->next = pt->next;
		else used.first = pt->next;
		if(pt->next == NULL) used.last = p;

		if(changed) {
		    if(ready.first == NULL) available = ready.first = ready.last = pt;
		    else {
			if(available == NULL) available = pt;
			ready.last->next = pt;
			ready.last = pt;
		    }
		    ready.last->next = NULL;
		    ready.last->in_use = true;
		    LOG_MSG("Object %d found used, moving to ready queue...", id);
		} else {
		    if(free.first == NULL) {
			free.first = free.last = pt;
			free.first->next = NULL;
		    } else {
			pt->next = free.first;
			free.first = pt;
		    }
		    LOG_MSG("Object %d found not used, moving back to free queue...", id);
		}
		MUTEX_UNLOCK(queue_lock);
		return true;
	    }
	    p = pt;
	    pt = pt->next;
	}

	LOG_MSG("Object %d not found!", id);
	MUTEX_UNLOCK(queue_lock);
	return false;
    }

    int GetUsed() {
	MUTEX_LOCK(queue_lock);
	if(available == NULL) {
	    LOG_MSG("GetUsed: ready queue empty or no objects available");
#ifdef DEBUG
	    ListQueues();
#endif
	    MUTEX_UNLOCK(queue_lock);
	    return 0x80000000;
	}

	int id = available->id;
	LOG_MSG("Returning %d from the ready queue", id);

	MUTEX_UNLOCK(queue_lock);
	return id;
    }

    bool Release(char id) {
	MUTEX_LOCK(queue_lock);
	LIST * pt = ready.first;

	LOG_MSG("Release: searching in ready queue for %d..", id);
	while(pt) {
	    if(pt->id == id) {
		pt->in_use = false;
		goto end;
	    }
	    pt = pt->next;
	}

	LOG_MSG("Release: searching in used queue for %d..", id);
	pt = used.first;
	while(pt) {
	    if(pt->id == id) {
		pt->in_use = false;
		goto end;
	    }
	    pt = pt->next;
	}

	MUTEX_UNLOCK(queue_lock);
	return false;
end:
	MUTEX_UNLOCK(queue_lock);
	return true;
    }

    void Flush() {
	MUTEX_LOCK(queue_lock);

	int i = 0;
	LIST * pt, * p;
	pt = free.first;
	while(pt) {
	    p = pt;
	    pt = pt->next;
	    delete p;
	};

	pt = used.first;
	while(pt) {
	    p = pt;
	    pt = pt->next;
	    delete p;
	    i++;
	};

	pt = ready.first;
	while(pt) {
	    p = pt;
	    pt = pt->next;
	    delete p;
	    i++;
	};

	used.first = used.last = NULL;
	free.first = free.last = NULL;
	ready.first = ready.last = NULL;
	available = &empty;

	LOG_MSG("Flushed %d entries", i);
	MUTEX_UNLOCK(queue_lock);
    }

    ~QUEUE() {
	Flush();
	MUTEX_DESTROY(queue_lock);
    }

};

#undef MODULE

#endif // FIFO_H