Jan Kończak

This is an old revision of the document!

Aside from shared memory, message passing is a way to synchronize threads and processes. Your assignment is to implement a basic message passing library for communicating threads within a process. You must implement the functions thread_init, thread_make, thread_term, thread_myid, thread_send, thread_mcast, thread_receive and thread_gather. Below, you are provided a header file with declarations of these functions and the description of how they have to work. You must not modify the header file. For thread management and synchronisation you must use POSIX API.

Your code will be run against automated tests verifying example scenarios. Code that does not compile, or does not implement the functions from the header file, or uses API for thread management and synchronisation other than POSIX API will be graded negatively. The solutions will be graded basing on the correctness of the code, use of fine-grained locking and correctness of memory management. Moreover, severe code illegibility, spaghetti code, code bloat and redundancy will lower the grade.

The deadline for the assignment is 02.06.2024 (AoE). Please send the solutions to me as an attachment to an e-mail message with the subject beginning with [OSCP].

header.h

#pragma once
#include <stddef.h>
 
/**
 * Placeholder struct for a message.
 * 
 * Each message internally has a "use count" (aka "reference count").
 * When the use count reaches zero, the memory of the message is freed.
 * 
 * YOU ARE SUPPOSED NOT TO IMPLEMENT THIS STRUCTURE
 */
struct thread_msg;
typedef struct thread_msg thread_msg;
 
/**
 * Increments the use count of a message.
 * 
 * YOU ARE SUPPOSED NOT TO IMPLEMENT THIS FUNCTION
 */
void thread_msg_inc_use_count(thread_msg* msg);
 
/**
 * Decrements the use count of a message and frees memory if the use
 * count reached zero.
 * 
 * YOU ARE SUPPOSED NOT TO IMPLEMENT THIS FUNCTION
 */
void thread_msg_dec_use_count(thread_msg* msg);
 
#ifndef THREAD_MSGQUEUE_SIZE
#define THREAD_MSGQUEUE_SIZE 8
#endif
 
/**
 * This function must be called before any other 'thread_*' function.
 * 
 * It may initialize data, but may as well be empty. It's up to you.
 */
void thread_init();
 
/**
 * Creates a thread and runs 'start_routine' with argument 'arg' in
 * the new thread.
 *
 * Returns a numeric identifier of the new thread.
 * The identifiers shall be consecutive numbers starting from 0.
 * Identifier 0 must always be the main thread.
 * 
 * Each thread must have a message queue in that at most 
 * 'THREAD_MSGQUEUE_SIZE' messages are stored.
 *
 * An attempt to queue a message when the queue is full must block
 * the thread that wants to queue the message until some message
 * gets dequeued.
 * 
 * If the 'start_routine' finishes, 'thread_term' must be automatically
 * called.
 */
int thread_make(void (*start_routine)(void*), void* arg);
 
/**
 * Terminates the current thread. Must free all associated memory,
 * including the memory of the messages from the queue by calling
 * 'thread_msg_dec_use_count' on each such message.
 * Must wake all processes waiting for messages from this thread.
 */
void thread_term();
 
/**
 * Returns the identifier of the current thread, identical to the one
 * returned by 'thread_make' which created the thread, or, for the main
 * thread, equal 0.
 */
int thread_myid();
 
/**
 * Puts a message 'msg' to the message queue of the thread identified
 * by 'id'.
 * 
 * Returns 0 on successful enqueueing the message. If this is the case,
 * the thread 'id' takes over the responsibility for managing the memory
 * of 'msg'.
 * 
 * On errors, returns -1 and sets ERRNO accordingly:
 *  - ESRCH when there is no thread 'id'
 *  - ECANCELED when the target thread terminated before the message has been queued
 */
int thread_send(int id, thread_msg* msg);
 
/**
 * Enqueues a message 'msg' to 'id_count' first thread identifiers
 * in the array 'ids'.
 * 
 * The threads indicated in the 'ids' array may not exist.
 * 
 * Returns the number of times the message has been queued.
 *
 * Must call 'thread_msg_inc_use_count' one time less than the number of
 * times the message has been queued.
 * 
 * │ Example:
 * │ | int ids[] = {2, 3, 4, 5, 6};
 * │ | thread_mcast(ids, 4, msg);
 * │ This will attempt to send 'msg' to threads 2,3,4,5 (notice id_count==4)
 * │ Provided that thread 5 no longer runs, the return value is '3' (for the
 * │ message has been put to the queue of threads 2,3,4) and executes the
 * │ 'thread_msg_inc_use_count' function two times (for now in total three
 * │ threads "use" the message).
 */
int thread_mcast(int ids[], size_t id_count, thread_msg* msg);
 
/**
 * If the queue for this thread is empty, waits until any message appears. 
 * Then, writes the first message from the queue to '*msg', writes its sender
 * identifier to '*id' and removes the message from the queue.
 */
void thread_receive(int * id, thread_msg** msg);
 
/**
 * Waits for a message from the threads identified by first 'id_count'
 * identifiers in 'ids' array.
 * 
 * Writes the messages to 'msg[0]', 'msg[1]', … , 'msg[id_count-1]' and
 * removes the messages from the queue.
 * 
 * If for some thread identifier amongst 'ids' there is neither a message
 * from this identifier nor there is no thread with such identifier, the
 * corresponding 'msg[id]' is set to NULL.
 * 
 * If identifiers in the 'ids' repeat, then multiple messages from the
 * repeating identifiers should be received.
 *
 * Returns the number of messages gathered.
 * 
 * │ Example:
 * │ | thread_msg* msgs[4];
 * │ | int ids[] = {2, 3, 4, 5};
 * │ | thread_gather(ids, 4, msgs);
 * │ Say that in the queue there is a message m₆ from thread 6 and m₃ from
 * │ thread 3, and threads 2 and 3 already terminated.
 * │ Then, thread_gather will wait until threads 4 and 5 do any of:
 * │ thread_send or thread_mcast to this thread or thread_term.
 * │ Say that thread 5 send m₅ and thread 4 mcasts m₄.
 * │ Once this happens, thread_gather writes {NULL, m₃, m₄, m₅} to msgs,
 * │ and returns '3'.
 * │ Obviously the messages m₃, m₄, m₅ are removed from the queue.
 * 
 * Warning: if the number of messages to be gathered is larger than the
 * size of the queue, this function must still work correctly without
 * any deadlock!
 */
int thread_gather(int ids[], size_t id_count, thread_msg *msgs[]);

An example code that uses the library is provided here:

#include "header.h"
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
 
struct thread_msg{
    atomic_int uc;
    char * txt;
};
 
thread_msg* thread_msg_make(const char * str);
 
void t1(void*_);
 
void t2(void* arg);
 
int main(){
    thread_init();
 
    // create two more threads
    int id1 = thread_make(t1, NULL);
    int id2 = thread_make(t2, (void*)(intptr_t)id1);
 
    // send two messages to the first thread 
    thread_msg* msg1 = thread_msg_make("FirstMessage");
    thread_send(id1, msg1);
    // the second message contains the ID of the second thread
    thread_msg* msg2 = thread_msg_make("X");
    sprintf(msg2->txt, "%d", id2);
    thread_send(id1, msg2);
 
    // send a message to the second thread
    thread_msg* msg3 = thread_msg_make("Hello ");
    thread_send(id2, msg3);
 
    // receive a message from the second thread
    int sender = id2;
    thread_msg* msg4;
    thread_gather(&sender, 1, &msg4);
    thread_msg_dec_use_count(msg4);
 
    return 0;
}
 
void t1(void*_){
    // receive a message and print its contents
    int sender;
    thread_msg * msg;
    thread_receive(&sender, &msg);
    printf("%s\n", msg->txt);
    thread_msg_dec_use_count(msg);
 
    // receive a message and extract a number (the ID of the second thread) from it
    thread_receive(&sender, &msg);
    int id2;
    sscanf(msg->txt, "%d", &id2);
    thread_msg_dec_use_count(msg);
 
    // send a message to two threads
    int tgt[]={0, id2};
    msg = thread_msg_make("world!");
    thread_mcast(tgt, 2, msg);
    // notice that thread 0 does not retreive the message from the queue
}
 
void t2(void* arg){
    // receive messages from the two threads and output them
    int senders[]={0, (intptr_t) arg};
    thread_msg* msgs[2];
    thread_gather(senders, 2, msgs);
    printf("%s%s\n", msgs[0]->txt, msgs[1]->txt);
    thread_msg_dec_use_count(msgs[0]);
    thread_msg_dec_use_count(msgs[1]);
 
    // send a message to the main thread
    thread_msg * msg = thread_msg_make(NULL);
    thread_send(0, msg);
 
    thread_term();
}
 
thread_msg* thread_msg_make(const char * str){
    thread_msg* msg = malloc(sizeof(thread_msg));
    if(str==NULL)
        msg->txt = NULL;
    else {
        msg->txt = malloc(strlen(str)+1);
        strcpy(msg->txt, str);
    }
    msg->uc=1;
    return msg;
}
 
void thread_msg_inc_use_count(thread_msg* msg){
    msg->uc++;
}
 
void thread_msg_dec_use_count(thread_msg* msg){
    if(!--msg->uc){
        free(msg->txt);
        free(msg);
    }
}