Aros/Developer/Docs/Libraries/Thread

Created by Rob Norris December 1st and 3rd 2007 to provide some threading possibilities to AROS to aid in bringing his port of webkit called Traveller. pthreads are not supported in AROS. Threads were introduced to provide a lower system expense over processes (tasks).

Author: Robert Norris <rob@cataclysm.cx> Last update: 2007-11-30

This is trivial library to provide basic threads and synchronisation primitives to higher-level libraries and applications. It currently provides the following:

Threads
Mutexes
Condition variables

This is a work-in-progress. Expect freakish things to happen if you use it. Please let me know if something doesn't work the way you want or expect it to. Both the interface and semantics are fluid, so you can get things changed if you want, but on the other hand you'll need to track changes to the library in your app.

This library is deliberately designed not to be POSIX threads. My hope is that it could be used to implement them, but it won't ever be that all by itself. POSIX thread semantics are more complicated than I want to deal with at this level.

The biggest omission at this stage is a way to force a thread to exit. Its hard to do because there's no way to know what resources the thread currently has open in order to close them. I have some ideas, but I don't know that it can be done well without proper task resource tracking.

In a similar vein, when your application or library closes thread.library (in an app, this will typically happen when main() exits), it will wait for any threads that are still running to finish. You'll get some warnings in the kernel log when this happens. The main process can't be allowed to exit while threads are still running as the main thread is usually holding resources that the threads need that will be deallocated when the thread exits, eg the program code itself. As long as the threads run in the same address space as the main process there's very little that can be done about this.

This also makes detached threads rather meaningless. The detached thread semantics are provided in the hope they can be fully implemented in the future.

TODO

A way to gracefully ask a thread to exit (giving the thread opportunity to clean up)
A way to wait for all threads to finish (including detached threads, so this can be used usefully to wait for threads before exiting in main()).
A true way to detach running threads such that they can exist after the main process exits.
Can exiting the main process while threads are still running be handled better?
Read/write ("promotable") mutexes, where you can request to obtain a lock exclusively while already holding it as a shared lock. This can avoid races in some situations.

What to do when the main task exits while threads are still running. There's a bunch of bad things that can happen, but the best thing is to simply detach the threads and allow them to continue running.

POSIX pthreads provide Thread Management, Mutex Variables (syncing), and Condition Variables (more syncing but depends on a mutex lock).

Library=

have to include , declare struct Library *ThreadBase, and open/close the library with OpenLibrary()/CloseLibrary(). you need proto/thread.h, which will include clib/thread_protos.h for you.

What is the current version number for thread.library (OpenLibrary () call) ? what's the current version, I just use 0.

If you use any socket stuff, don't forget to open bsdsocket.library too.

CreateThread() CurrentThread() DetachThread() WaitThread() WaitAllThreads()

uint32_t id = CreateThread(entry, data);
if (id < 0)
    printf("thread creation failed\n");
else
    printf("thread %d created\n", id);

void *ret;
WaitThread(id, &ret);

#include <libraries/thread.h>
#include <proto/thread.h>
#include <proto/dos.h>
#include <stdio.h>
#include <stdint.h>

void *thread_main(void *data) {
    ThreadIdentifier id = CurrentThread();
    int i;

    printf("[%d] starting\n", id);

    for (i = 0; i < 10; i++) {
        printf("[%d] count: %d\n", id, i);
        Delay(25);
    }

    printf("[%d] exiting\n", id);

    return NULL;
}

int main (int argc, char **argv) {
    ThreadIdentifier t1, t2;

    t1 = CreateThread(thread_main, NULL);
    printf("created thread %d\n", t1);

    Delay(100);

    t2 = CreateThread(thread_main, NULL);
    printf("created thread %d\n", t2);

    printf("waiting for thread %d\n", t2);
    WaitThread(t2, NULL);
    printf("thread %d completed\n", t2);

    printf("waiting for thread %d\n", t1);
    WaitThread(t1, NULL);
    printf("thread %d completed\n", t1);

    return 0;
}

#include <libraries/thread.h>
#include <proto/thread.h>
#include <proto/dos.h>
#include <stdio.h>

void *thread_main(void *data) {
    ThreadIdentifier id = CurrentThread();

    printf("[%d] starting\n", id);

    Delay(50);

    printf("[%d] exiting\n", id);

    return (void *) id;
}

int main (int argc, char **argv) {
    int i;
    ThreadIdentifier id[10], ret;

    for (i = 0; i < 10; i++) {
        id[i] = CreateThread(thread_main, NULL);
        printf("created thread %d\n", id[i]);
        Delay(25);
    }

    for (i = 0; i < 10; i++) {
        printf("waiting for thread %d\n", id[i]);
        WaitThread(id[i], (void **) &ret);
        printf("thread %d return %d\n", id[i], ret);
    }

    return 0;
}

CreateMutex() DestroyMutex() LockMutex() TryLockMutex() UnlockMutex()

Mutex mutex = CreateMutex();

LockMutex(mutex);
WaitCondition(cond, mutex);
UnlockMutex(mutex);

#include <libraries/thread.h>
#include <proto/thread.h>
#include <proto/dos.h>
#include <stdio.h>
#include <stdint.h>

void *locker_thread(void *data) {
    Mutex mutex = (Mutex) data;
    ThreadIdentifier id = CurrentThread();

    printf("[%d] starting, locking the mutex\n", id);
    LockMutex(mutex);

    printf("[%d] got it, pausing for 5s\n", id);
    Delay(250);

    printf("[%d] unlocking the mutex\n", id);
    UnlockMutex(mutex);

    printf("[%d] all done, exiting\n", id);

    return NULL;
}

void *waiter_thread(void *data) {
    Mutex mutex = (Mutex) data;
    ThreadIdentifier id = CurrentThread();

    printf("[%d] starting, locking the mutex\n", id);
    LockMutex(mutex);

    printf("[%d] got it, unlocking\n", id);
    UnlockMutex(mutex);

    printf("[%d] all done, exiting\n", id);

    return NULL;
}

int main (int argc, char **argv) {
    Mutex mutex;
    ThreadIdentifier tl, tw;

    printf("creating mutex\n");
    mutex = CreateMutex();

    printf("starting locker thread\n");
    tl = CreateThread(locker_thread, (void *) mutex);

    printf("sleeping for 2s\n");
    Delay(100);

    printf("starting waiter thread\n");
    tw = CreateThread(waiter_thread, (void *) mutex);

    printf("waiting for locker thread to exit\n");
    WaitThread(tl, NULL);

    printf("waiting for waiter thread to exit\n");
    WaitThread(tw, NULL);

    printf("destroying the mutex\n");
    DestroyMutex(mutex);

    printf("all done\n");

    return 0;
}

BroadcastCondition() CreateCondition() DestroyCondition() WaitCondition() SignalCondition()

#include <exec/memory.h>
#include <libraries/thread.h>
#include <proto/exec.h>
#include <proto/dos.h>
#include <proto/thread.h>
#include <stdio.h>
#include <stdint.h>

struct thread_data {
    Mutex mutex;
    Condition cond;
};

void *waiter_thread(void *data) {
    struct thread_data *td = (struct thread_data *) data;
    ThreadIdentifier id = CurrentThread();

    printf("[%d] starting, locking the mutex\n", id);
    LockMutex(td->mutex);

    printf("[%d] waiting on the condition\n", id);
    WaitCondition(td->cond, td->mutex);

    printf("[%d] condition signalled, unlocking the mutex\n", id);
    UnlockMutex(td->mutex);

    printf("[%d] all done, exiting\n", id);

    return NULL;
}

int main (int argc, char **argv) {
    struct thread_data *td;
    int i;

    td = AllocMem(sizeof(struct thread_data), MEMF_PUBLIC | MEMF_CLEAR);

    printf("creating mutex\n");
    td->mutex = CreateMutex();

    printf("creating condition\n");
    td->cond = CreateCondition();

    printf("starting waiter threads\n");
    for (i = 0; i < 5; i++)
        CreateThread(waiter_thread, (void *) td);

    printf("sleeping for 2s\n");
    Delay(100);

    printf("signalling condition\n");
    SignalCondition(td->cond);

    printf("sleeping for 2s\n");
    Delay(100);

    printf("broadcasting condition\n");
    BroadcastCondition(td->cond);

    printf("waiting for threads to exit\n");
    WaitAllThreads();

    printf("destroying the condition\n");
    DestroyCondition(td->cond);

    printf("destroying the mutex\n");
    DestroyMutex(td->mutex);

    FreeMem(td, sizeof(struct thread_data));

    printf("all done\n");

    return 0;
}

#include <exec/memory.h>
#include <libraries/thread.h>
#include <proto/exec.h>
#include <proto/dos.h>
#include <proto/thread.h>
#include <stdio.h>
#include <stdint.h>

struct thread_data {
    Mutex mutex;
    Condition cond;
};

void *waiter_thread(void *data) {
    struct thread_data *td = (struct thread_data *) data;
    ThreadIdentifier id = CurrentThread();

    printf("[%d] starting, locking the mutex\n", id);
    LockMutex(td->mutex);

    printf("[%d] waiting on the condition\n", id);
    WaitCondition(td->cond, td->mutex);

    printf("[%d] condition signalled, unlocking the mutex\n", id);
    UnlockMutex(td->mutex);

    printf("[%d] all done, exiting\n", id);

    return NULL;
}

int main (int argc, char **argv) {
    struct thread_data *td;
    ThreadIdentifier tw;

    td = AllocMem(sizeof(struct thread_data), MEMF_PUBLIC | MEMF_CLEAR);

    printf("creating mutex\n");
    td->mutex = CreateMutex();

    printf("creating condition\n");
    td->cond = CreateCondition();

    printf("starting waiter thread\n");
    tw = CreateThread(waiter_thread, (void *) td);

    printf("sleeping for 2s\n");
    Delay(100);

    printf("signalling condition\n");
    SignalCondition(td->cond);

    printf("waiting for waiter thread\n");
    WaitThread(tw, NULL);

    printf("destroying the condition\n");
    DestroyCondition(td->cond);

    printf("destroying the mutex\n");
    DestroyMutex(td->mutex);

    FreeMem(td, sizeof(struct thread_data));

    printf("all done\n");

    return 0;
}

Examples

An alternative (not used in AROS) of implementing threads.

/* sys_thrad.h */
struct SysThread;
struct SysMutex;

enum SysThreadPriority
{
	SYSTHREAD_PRIORITY_LOW,
	SYSTHREAD_PRIORITY_NORMAL,
	SYSTHREAD_PRIORITY_HIGH,
};

struct SysThread *Sys_Thread_CreateThread(void (*entrypoint)(void *), void *argument);
void Sys_Thread_DeleteThread(struct SysThread *thread);
int Sys_Thread_SetThreadPriority(struct SysThread *thread, enum SysThreadPriority priority);

struct SysMutex *Sys_Thread_CreateMutex(void);
void Sys_Thread_DeleteMutex(struct SysMutex *mutex);
void Sys_Thread_LockMutex(struct SysMutex *mutex);
void Sys_Thread_UnlockMutex(struct SysMutex *mutex);

/*
Copyright (C) 2008-2011 Mark Olsen

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 <exec/semaphores.h>
#include <dos/dos.h>
#include <dos/dostags.h>

#include <proto/exec.h>
#include <proto/dos.h>

#include "sys_thread.h"

struct SysThread
{
	struct MsgPort *msgport;
	struct Message msg;
	struct Process *process;

	void (*entrypoint)(void *);
	void *argument;
};

struct SysMutex
{
	struct SignalSemaphore sem;
};

static void Sys_Thread_Trampoline()
{
	struct SysThread *thread;
	
	thread = FindTask(0)->tc_UserData;

	thread->entrypoint(thread->argument);
	
	Forbid();
	ReplyMsg(&thread->msg);
}

struct SysThread *Sys_Thread_CreateThread(void (*entrypoint)(void *), void *argument)
{
	struct SysThread *thread;

	thread = AllocVec(sizeof(*thread), MEMF_ANY);
	if (thread)
	{
		thread->msgport = CreateMsgPort();
		if (thread->msgport)
		{
			thread->msg.mn_Node.ln_Type = NT_MESSAGE;
			thread->msg.mn_ReplyPort = thread->msgport;
			thread->msg.mn_Length = sizeof(thread->msg);

			thread->entrypoint = entrypoint;
			thread->argument = argument;

			thread->process = CreateNewProcTags(NP_Entry, Sys_Thread_Trampoline,
			                                    NP_UserData, thread,
			                                    NP_Name, "Fodquake Thread",
			                                    NP_StackSize, 32768,
			                                    TAG_DONE);
			if (thread->process)
			{
				return thread;
			}

			DeleteMsgPort(thread->msgport);
		}
		FreeVec(thread);
	}

	return 0;
}

void Sys_Thread_DeleteThread(struct SysThread *thread)
{
	SetTaskPri(&thread->process->pr_Task, 0);

	while(!GetMsg(thread->msgport))
		WaitPort(thread->msgport);

	DeleteMsgPort(thread->msgport);
	FreeVec(thread);
}

int Sys_Thread_SetThreadPriority(struct SysThread *thread, enum SysThreadPriority priority)
{
	int pri;

	switch(priority)
	{
		case SYSTHREAD_PRIORITY_LOW:
			pri = -1;
			break;
		case SYSTHREAD_PRIORITY_HIGH:
			pri = 4;
			break;
		default:
			pri = 0;
			break;
	}

	SetTaskPri(&thread->process->pr_Task, pri);

	return 0;
}

struct SysMutex *Sys_Thread_CreateMutex(void)
{
	struct SysMutex *mutex;

	mutex = AllocVec(sizeof(*mutex), MEMF_ANY);
	if (mutex)
	{
		InitSemaphore(&mutex->sem);

		return mutex;
	}

	return 0;
}

void Sys_Thread_DeleteMutex(struct SysMutex *mutex)
{
	FreeVec(mutex);
}

void Sys_Thread_LockMutex(struct SysMutex *mutex)
{
	ObtainSemaphore(&mutex->sem);
}

void Sys_Thread_UnlockMutex(struct SysMutex *mutex)
{
	ReleaseSemaphore(&mutex->sem);
}

References

uint32_t CreateThread(ThreadEntryFunction entry, void *data) 
BOOL WaitThread(uint32_t thread_id, void **result) 
void WaitAllThreads() 
BOOL DetachThread(uint32_t thread_id) 
uint32_t CurrentThread() 
void *CreateMutex() 
BOOL DestroyMutex(void *mutex) 
void LockMutex(void *mutex) 
BOOL TryLockMutex(void *mutex) 
void UnlockMutex(void *mutex) 
void *CreateCondition() 
BOOL DestroyCondition(void *cond) 
BOOL WaitCondition(void *cond, void *mutex) 
void SignalCondition(void *cond) 
void BroadcastCondition(void *cond) 
void ExitThread(void *result)