job.c

mysee网络直播源代码Mysee Lite是Mysee独立研发的网络视频流媒体播放系统。在应有了P2P技术和一系列先进流媒体技术之后

/*
 *  Openmysee
 *
 *  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
 *
 */
				 
#include "echo.h"
struct JobDes *JobCacheHead;
static int JobCacheCount;
extern int JobHighWater;

inline struct JobDes *newJob ()
{
	struct JobDes *p = JobCacheHead;
	if (p)
	{
		JobCacheHead = p->next;
		JobCacheCount --;
		p->start = p->len = 0;
		p->pc = NULL;
		p->p = NULL;
	}
	else
		p = calloc (1, sizeof (struct JobDes));
	p->blockid = -1;
	return p;
}

void inline setblockId (struct JobDes *pj, int id)
{
	pj->blockid = id;
}

void freeJobCache ()
{
	struct JobDes *p, *nextp;
	for (p=JobCacheHead; p; p=nextp)
	{
		nextp = p->next;
		free (p);
	}
	JobCacheHead = NULL;
	JobCacheCount = 0;
}

static struct JobDes *freeJob (struct JobDes *head, struct JobDes *p)
{
	if (p == NULL) return head;
	if (p->pc) p->pc->numjob --;
	p->p->numjob --;
	if (head == p)
		head = p->next;
	if (head == p)
		head = NULL;
	p->prev->next = p->next;
	p->next->prev = p->prev;
	if (JobCacheCount > JobHighWater)
	{
		free (p);
	} else
	{
		p->next = JobCacheHead;
		JobCacheHead = p;
		JobCacheCount ++;
	}
	return head;
}

inline struct JobDes *findEnoughBuffer (struct Session *q, struct Channel *pc, int size)
{
	struct JobDes *p = q->head;
	if (p == NULL) return NULL;
	do
	{
		if (p->pc == pc && p->blockid == -1 && MAX_MSG_SIZE - p->len - p->start > size)
			return p;
		p = p->next;
	} while (p!=q->head);
	return NULL;
}

inline char *getJobBuffer (struct JobDes *p, int *max)
{
	*max = MAX_MSG_SIZE - p->len - p->start;
	return p->buffer+p->start+p->len;
}

void addJob (struct Session *q, struct Channel *pc, struct JobDes *n)
{
	n->p = q;
	n->pc = pc;
	if (q->head == NULL)
	{
		q->head = n;
		n->next = n;
		n->prev = n;
	} else
	{
		n->next = q->head;
		n->prev = q->head->prev;
		n->next->prev = n;
		n->prev->next = n;
	}
	q->numjob ++;
	if (pc) pc->numjob ++;
}

void deleteJob (struct Session *q, struct Channel *pc, int *ids, int num)
{
	int i;
	struct JobDes *p, *nextp;

	if ((p=q->head) == NULL) return;
	for (;;)
	{
		nextp = p->next;
		if (p->pc == pc)
		{
			for (i=0; i<num; i++)
			{
				if (ids[i] == p->blockid)
				{
					q->head = freeJob (q->head, p);
					break;
				}
			}
		}
		if (q->head == NULL || q->head == nextp)
			break;
		p = nextp;
	}
}

void deleteAll (struct Session *q)
{
	struct JobDes *p, *nextp;

	if ((p=q->head) == NULL) return;
	for (;;)
	{
		nextp = p->next;
		q->head = freeJob (q->head, p);
		if (q->head == NULL)
			break;
		p = nextp;
	}
}

void deleteChannel (struct Session *q, struct Channel *pc)
{
	struct JobDes *p, *nextp;

	if ((p=q->head) == NULL) return;
	for (;;)
	{
		nextp = p->next;
		if (p->pc == pc)
			q->head = freeJob (q->head, p);
		if (q->head == NULL || q->head == nextp)
			break;
		p = nextp;
	}
}


int processJobs (struct Session *q)
{
	struct JobDes *p, *nextp;
	unsigned int total = 0, this_write;
	if ((p=q->head) == NULL) return 0;
	for (;;)
	{
		nextp = p->next;
		if ((this_write = write (q->socket, p->buffer+p->start, p->len))
				<= 0)
		{
			PDEBUG ("Error in write in session %d.\n", q->socket);
			return -1;
		}
		p->len -= this_write;
		p->start += this_write;
		total += this_write;
		if (p->len == 0)
			q->head = freeJob (q->head, p);
		else
			break;
		if (q->head == NULL)
			break;
		p = nextp;
	}
	return total;
}

#ifdef __TEST_JOB
main ()
{
	struct Session *q = calloc (sizeof (*q), 1);
	struct Channel *qc = calloc (sizeof (*qc), 1);
	struct JobDes *p;
	int array[2];
	assert (q);
	p = newJob ();
	if (p == NULL || p->blockid != -1 || p->pc != NULL || p->start != 0
			|| p->len != 0 || p->p != NULL)
		assert (0);
	setblockId (p, 10);
	if (p->blockid != 10)
		assert (0);
	addJob (q, qc, p);

	// a new job
	p = newJob ();
	if (p == NULL || p->blockid != -1 || p->pc != NULL || p->start != 0
			|| p->len != 0 || p->p != NULL)
		assert (0);
	setblockId (p, 20);
	if (p->blockid != 20)
		assert (0);
	addJob (q, NULL, p);

	// a new job
	p = newJob ();
	if (p == NULL || p->blockid != -1 || p->pc != NULL || p->start != 0
			|| p->len != 0 || p->p != NULL)
		assert (0);
	setblockId (p, 30);
	if (p->blockid != 30)
		assert (0);
	addJob (q, qc, p);

	// a new job
	p = newJob ();
	if (p == NULL || p->blockid != -1 || p->pc != NULL || p->start != 0
			|| p->len != 0 || p->p != NULL)
		assert (0);
	setblockId (p, 40);
	if (p->blockid != 40)
		assert (0);
	addJob (q, NULL, p);


	// a new job
	p = newJob ();
	if (p == NULL || p->blockid != -1 || p->pc != NULL || p->start != 0
			|| p->len != 0 || p->p != NULL)
		assert (0);
	setblockId (p, 50);
	if (p->blockid != 50)
		assert (0);
	addJob (q, qc, p);

	if (q->head == NULL || q->head->blockid != 10 || q->head->next == NULL
			|| q->head->next->blockid != 20
			|| q->head->next->next == NULL
			|| q->head->next->next->blockid != 30
			|| q->head->next->next->next == NULL
			|| q->head->next->next->next->blockid != 40
			|| q->head->next->next->next->next == NULL
			|| q->head->next->next->next->next->blockid != 50
			|| q->head->next->next->next->next->next == NULL
			|| q->head->next->next->next->next->next->blockid != 10)
		assert (0);
	array[0] = 10;
	array[1] = 20;
	deleteJob (q, qc, array, 2);
	if (q->head == NULL || q->head->blockid != 20 || q->head->next == NULL
			|| q->head->next->blockid != 30
			|| q->head->next->next == NULL
			|| q->head->next->next->blockid != 40
			|| q->head->next->next->next == NULL
			|| q->head->next->next->next->blockid != 50
			|| q->head->next->next->next->next == NULL
			|| q->head->next->next->next->next->blockid != 20)
		assert (0);
	deleteChannel (q, qc);
	if (q->head == NULL || q->head->blockid != 20 || q->head->next == NULL
			|| q->head->next->blockid != 40)
		assert (0);
	deleteAll (q);
	if (q->head != NULL)
		assert (0);
	free (q);
	free (qc);
	freeJobCache ();
}
#endif