Miscellany

////=== wqueue.h ===//#ifndef WQUEUE_H_#define WQUEUE_H_ 1#include <pthread.h>/*Definition af arbejdsfunktionen.*/typedef void (*work_f)(void *);typedef struct wnode {	struct wnode *prev;	struct wnode *next;	work_f funk;	void *data;	unsigned int pri;} wnode_t;/*Åben definition af en arbejdskø instans. Denne kanændress efter behov, sålænge det ikke ændrer på nedenståendefunktioners grænseflade.*/typedef struct wqueue {	struct wnode *first;	struct wnode *last;	int threads;	int idlers;	pthread_mutex_t condition_mutex;	pthread_cond_t condition_cond;} wqueue_t;typedef struct qthread {	pthread_t thread;	struct wqueue *wq;	int id;} qthread_t;/*Indsætter et stykke arbejde i form af en funktion 'func' og detdata 'data' den skal kaldes med. Prioriteten 'pri' angiverarbejdets prioritet. 'wq' peger på en instans af arbejdskøen.Lykkedes indsættelsen skal funktionen returnerer '0'. Ved fejl '-1'.*/int wqueue_insert(wqueue_t *wq, unsigned int pri, work_f func, void *data);/*Udfører et stykke arbejde fra køen.Hvis køen er tom returneres '0', ellers '1'.*/int wqueue_run(wqueue_t *wq);/*Trådsikker (thread-safe) udgave af 'wqueue_insert'.*/int wqueue_ts_insert(wqueue_t *wq, unsigned int pri, work_f func, void *data);/*Starter en trådpulje (thread pool) med 'cnt' tråde. Trådene udtager arbjedefra køen og udfører det. Hvis der ikke er noget arbejde skal trådende blokkereog vente på at der kommer nyt arbejde.Når køen er tom og alle tråde er færdige med at arbejde returneres '0'. Ved fejlreturneres '-1';*/int wqueue_thread_pool(wqueue_t *wq, unsigned int cnt);#endif//// === wqueue.c ===//#include <stdio.h>#include <stdlib.h>#include <pthread.h>#include "wqueue.h"int wqueue_insert(wqueue_t *wq, unsigned int pri, work_f func, void *data) {	wnode_t* n = malloc(sizeof(wnode_t));	if(n==NULL)		return -1;	n->pri = pri;	n->next = NULL;	n->prev = NULL;	n->funk = func;	n->data = data;	if(wq->first==NULL) {		wq->first = n;		wq->last = n;	} else {		wnode_t* tmp = (wnode_t*)malloc(sizeof(wnode_t));		if(tmp==NULL)			return -1;		tmp = wq->first;		while(tmp->pri >= n->pri) {			tmp = tmp->next;			if(tmp==NULL)				break;		}		if(tmp==wq->first) {			wq->first = n;			n->next = tmp;			tmp->prev = n;		} else if(tmp==NULL) {			tmp = wq->last;			wq->last = n;			n->prev = tmp;			tmp->next = n;		} else {			n->next = tmp;			n->prev = tmp->prev;			n->prev->next = n;			tmp->prev = n;		}	}	return 0;}int wqueue_run(wqueue_t *wq) {	wnode_t* tmp = malloc(sizeof(wnode_t));	if(wq->first==NULL)		return 0;	tmp = wq->first;	tmp->funk(tmp->data);	wq->first = tmp->next;	tmp->prev = NULL;	free(tmp);	return 1;}int wqueue_ts_insert(wqueue_t *wq, unsigned int pri, work_f func, void *data) {	printf("Inserting job with priroity %d", pri);	pthread_mutex_lock(&wq->condition_mutex);	printf(".");	wqueue_insert(wq, pri, func, data);	printf(".");	pthread_cond_signal(&wq->condition_cond);	printf(".");	pthread_mutex_unlock(&wq->condition_mutex);	printf(" done!\n");}void *finish_the_scene(void *vthread) {	qthread_t *thread = (qthread_t*)vthread;	wqueue_t *wq = (wqueue_t*)thread->wq;	wnode_t *tmp;	printf("Thread %d locks the mutex...\n", thread->id);	pthread_mutex_lock(&wq->condition_mutex);	printf("Thread %d begins the loop...\n", thread->id);	for(;;) {		if(wq->first==NULL)			break;		wq->idlers--;		tmp = wq->first;		wq->first = tmp->next;		tmp->prev = NULL;		pthread_cond_signal(&wq->condition_cond);		pthread_mutex_unlock(&wq->condition_mutex);		printf("Thread %d is executing function...\n", thread->id);		tmp->funk(tmp->data);		free(tmp);		pthread_mutex_lock(&wq->condition_mutex);		wq->idlers++;		while(wq->idlers==0) {			pthread_cond_wait(&wq->condition_cond, &wq->condition_mutex);		}		sleep(1);	}	pthread_mutex_unlock(&wq->condition_mutex);	printf("Thread %d freed!\n", thread->id);}int wqueue_thread_pool(wqueue_t *wq, unsigned int cnt) {	qthread_t *pthreads[cnt];	wnode_t *tmp;	unsigned int i;	tmp = wq->first;	wq->threads = cnt;	wq->idlers = cnt;	printf("Starting pool...\n");//	pthread_mutex_lock(&wq->condition_mutex);//	while(wq->first==NULL) {//		pthread_cond_wait(&wq->condition_cond, &wq->condition_mutex);//	}//	pthread_mutex_unlock(&wq->condition_mutex);	for(i=0; i<cnt; i++) {		pthreads[i] = malloc(sizeof(qthread_t));		pthreads[i]->id = i;		printf("Creating thread %d...", pthreads[i]->id);		pthread_create(&pthreads[i]->thread, NULL, finish_the_scene, (void *)pthreads[i]);		printf(" done!\n");	}	printf("All threads created.\n");	for(i=0; i<cnt; i++) {		printf("Launching thread %d...", pthreads[i]->id);		fflush(stdout);		pthread_join(pthreads[i]->thread, NULL);		printf(" done!\n");		fflush(stdout);	}	printf("I'm done!\n");	for(i=0; i<cnt; i++) {		free(pthreads[i]);	}	pthread_exit(NULL);	return 0;}//// === wqueue_run.c ===//#include <stdlib.h>#include <stdio.h>#include <pthread.h>#include "wqueue.h"wqueue_t *wq;void print_me(void *data) {	printf(":: %s\n", (char*)data);}void sum(void *data) {	int *a = (int*)data;	char *buffer = malloc(10*sizeof(char));	sprintf(buffer, "%i", a[0]+a[1]);	printf("Inserting subjobs...\n");	wqueue_ts_insert(wq, -1, &print_me, buffer);	wqueue_ts_insert(wq, -1, &free, buffer);}int main() {	wq = malloc(sizeof(wqueue_t));	if(wq==NULL)		return 1;	wq->first = NULL;	wq->last = NULL;	pthread_mutex_init(&wq->condition_mutex, NULL);	int p[2] = {5, 10};	int q[2] = {50, 100};	wqueue_ts_insert(wq, 3, (work_f)print_me, "Thread A. 3");	wqueue_ts_insert(wq, 2, (work_f)print_me, "Thread B. 2");	wqueue_ts_insert(wq, 3, (work_f)print_me, "Thread C. 3");	wqueue_ts_insert(wq, 4, (work_f)print_me, "Thread D. 4");	wqueue_ts_insert(wq, 3, (work_f)print_me, "Thread E. 3");	wqueue_ts_insert(wq, 2, (work_f)print_me, "Thread F. 2");	wqueue_ts_insert(wq, 1, (work_f)print_me, "Thread G. 1");	wqueue_ts_insert(wq, 3, (work_f)print_me, "Thread H. 3");	wqueue_ts_insert(wq, 1, (work_f)sum, q);	wqueue_ts_insert(wq, 4, (work_f)sum, p);	wqueue_thread_pool(wq, 3);	return 0;}