posix_thread_pool.c

#include <stdlib.h>
#include <stdbool.h>
#include <stdio.h>
#include <pthread.h>
#include <assert.h>

// ThreadSafeQueue
// ThreadPool

// 1. How to implement virtual functions
// 2. godbolt.org/com
// 3. https://github.com/radareorg/radare2

// class Mlass: public IMlass
// {
// 	virtual void f() override
// 	{
// 	}
// 
// 	void h()
// 	{
// 	}
// };
// 
// void f()
// {
// }
// 
// void g()
// {
// 	Mlass* m = new Mlass;
// 	for (int i = 0; i < 100000000; i++)
// 	{
// 		// 1. m->f(); this->[vptr+1]()
// 		// 3. m->h();
// 		// 2. f();
// 	}
// }

typedef void* (*fn)(void*);
typedef struct fn_bind
{
	fn f;
	void* arg;
} fn_bind;

//========================================
// Queue implementation via linked list
// Represents linked list node

// TODO: Make thread safe queue data structure using doubly linked-list

typedef struct node
{
	struct node* prev;
	struct node* next;
	void* value; // should fn_bind
} node;

void init_node(node** n)
{
	(*n) = (node*)malloc(sizeof(node));
	(*n)->next = NULL;
	(*n)->prev = NULL;
	(*n)->value = NULL;
}

void free_node(node* n)
{
	// TODO: Implement free(a.k.a. destructor) for node
}

// Constructor/Destructor
typedef struct queue
{
	node* head;
	node* tail;
	// TODO: pthread_mutex_t mutex;
} queue;

// Acts as a constructor
void init_queue(queue** q)
{
	*q = (queue*)malloc(sizeof(queue));
	// TODO: Init mutex
}

void push_queue(queue* q, void* arg)
{
	assert(q != NULL);
	assert(arg != NULL);

	// pthread_mutex_lock(q->mutex);

	if (q->tail == NULL)
	{
		init_node(&(q->tail));	

		q->head = q->tail;
		q->tail->value = arg;
		q->tail->prev = q->head;
		q->head->next = q->tail;
	}
	else
	{
		node* n;
		init_node(&n);
		n->value = arg;

		q->tail->next = n;
		n->prev = q->tail;
		n->next = NULL;
		q->tail = n;
	}
}

node* pop_queue(queue* q)
{
	// TODO: Implement pop logic for queue via doubly-linked list
	return NULL;
}

// Acts as a destructor
void free_queue(queue* q)
{
	// TODO: Implement free logic for queue (e.g. free head/tail using their destructor a.k.a. free_node)
	free(q);
}

//========================================

void* hello(void* arg)
{
	char* str = (char*)arg;
	printf("%s\n", str);
	return NULL;
}

void f(fn_bind* fb)
{
	fb->f(fb->arg);
}

typedef struct Counter
{
	pthread_mutex_t mutex;
	int counter;
} Counter;

void init_counter(Counter* c)
{
	pthread_mutex_init(&(c->mutex), NULL);
	c->counter = 0;
}

void incr_counter(Counter* c, int v)
{
	pthread_mutex_lock(&(c->mutex));
	printf("v=%d\n", v);
	c->counter += v;
	pthread_mutex_unlock(&(c->mutex));
}

void* routine(void* vc)
{
	Counter* c = (Counter*)vc;
	for (int i = 0; i < 3; i++)
	{
		incr_counter(c, i);
	}

	return NULL;
}

#define THREAD_COUNT 2
struct thread_pool
{
	pthread_t m_threads[THREAD_COUNT];	
	queue* m_queue;

	// TODO: Move to separate struct e.g. thread_safe_bool
	pthread_mutex_t m_stop_flag_mutex;
	bool m_stop;
};

void thread_worker(void* vthread_pool)
{
	thread_pool* tp = (thread_pool*)vthread_pool;
	while (true)
	{
		// TODO: 1. Check if queue is empty && m_stop flag is set exit
		// TODO: 2. Pop from queue and execute
	}
}

void init_thread_pool(thread_pool** tp)
{
	// TODO: Thread pool initialization
	// TODO: Malloc for tp
	for (size_t i = 0; i < THREAD_COUNT; i++)
	{
		pthread_create(&m_threads[i], NULL, thread_worker, tp); 
	}
}

void stop_queue()
{
	// TODO: 1. Lock stop flag mutex
	// TODO: 2. Set flag
}

void free_thread_pool(thread_pool* tp)
{
	// TODO: 1. Join threads
	// TODO: 2. Free queue e.g. call free_queue()
}

int main()
{
	// ==================================================
	// Example on thread usage with thread safe counter
	// Counter* c = (Counter*)malloc(sizeof(Counter));
	// init_counter(c);

	// pthread_t p1;
	// pthread_t p2;
	// pthread_create(&p1, NULL, routine, c);
	// pthread_create(&p2, NULL, routine, c);
	// pthread_join(p1, NULL);
	// pthread_join(p2, NULL);
	// printf("counter=%d\n", c->counter);
	// ==================================================

	// ==================================================
	// Example of function arguments binding in C
	// fn_bind fb;
	// fb.f = hello;
	// char* str = "Hello\n";
	// fb.arg = str;

	// f(&fb);
	// ==================================================

	// ==================================================
	// TODO: Client code e.g. some sort of producer
	size_t jobCount = 1024;
	queue* q;
	init_queue(&q);
	thread_pool* tp;
	init_thread_pool(&tp, q);
	for (size_t i = 0; i < jobCount; i++)
	{
		// TODO: Create fn_bind e.g. fb
		push_queue(q, fb);
	}

	stop_thread_pool(tp);
	free_thread_pool(tp);
	// ==================================================

	// ==================================================
	// Test code for queue
	// queue* q = NULL;
	// init_queue(&q);

	// node n;
	// n.value = str;
	// push_queue(q, &n);

	// TODO: Write producer and consumer threads. Make sure that the queue is thread safe.
	//			 Producer thread should wrap functions using 'fn_bind' structure.
	//			 Consumer thread should pop end execute wrappers.

	// node* tmp = q->tail->prev;
	// while (tmp != NULL)
	// {
	// 	printf("%x\n", tmp->value);
	// 	tmp = tmp->next;
	// }

	// free_queue(q);
	// ==================================================
	return 0;
}