Mutex

#include <stdlib.h> /*malloc, free*/
#include <stdio.h> /*printf*/
#include <ucontext.h> /*context functionality*/
#include <assert.h> /*assert*/
#include <signal.h>
#include <sys/time.h>
#include "green.h"

#define PERIOD 100
#define FALSE 0
#define TRUE 1
#define STACK_SIZE 4096
#define MAX 1000000

static ucontext_t main_cntx = {0};
static green_t main_green = {&main_cntx, NULL, NULL, NULL, NULL, FALSE};
static green_t* running = &main_green;
static sigset_t block;

static void init () __attribute__ ((constructor));
void timer_handler(int sig);

void init () {
    getcontext (&main_cntx);
    sigemptyset(&block);
    sigaddset(&block, SIGVTALRM);

    struct sigaction act = {0};
    struct timeval interval;
    struct itimerval period;

    act.sa_handler = timer_handler;
    assert(sigaction(SIGVTALRM, &act, NULL) == 0);

    interval.tv_sec = 0;
    interval.tv_usec = PERIOD;
    period.it_interval = interval;
    period.it_value = interval;
    setitimer(ITIMER_VIRTUAL, &period, NULL);
}

struct Queue {
    green_t* front;
    green_t* end;
};
struct Queue* queue;

// Function to add a key to queue
void myEnQueue(green_t* new) {

    if (queue->end == NULL) {
       queue->front = queue->end = new;
       return;
    }
    queue->end->next = new;
    queue->end = new;
}

// Function to remove a key from queue
struct green_t* myDeQueue() {
    if (queue->front == NULL) {
       return NULL;
    }
    //printf("1: queue->front = %p \n", queue->front);
    struct green_t* temp = queue->front;
    //temp->next = NULL;
    queue->front = queue->front->next;
    temp->next = NULL;
    if (queue->front == NULL) {
       queue->end = NULL;
    } else {
        //printf("2: queue->front = %p \n", queue->front);
    }
    return temp;
}

void removeFromReadyList(green_t* remove) {
    struct green_t* temp = queue->front;
    struct green_t* afterTemp = temp->next;
    if (temp == remove) {
        queue->front = queue->front->next;
        //printf("if (temp == remove), queue->front = %p \n", queue->front);
        return;
    }
    while (afterTemp != remove) {
        temp = temp->next;
        afterTemp = afterTemp->next;
    }
    if (afterTemp != NULL) {
        temp->next = afterTemp->next;
    }
}

int green_mutex_init(green_mutex_t *mutex) {
    mutex->taken = FALSE;
    mutex->susp = NULL;
}

int green_mutex_lock(green_mutex_t *mutex) {
    // block timer interrupt
    sigprocmask(SIG_BLOCK, &block, NULL);

    green_t *susp = running;
    while(mutex->taken) {
        // suspend the running thread. Should I have a susp list?
        // add susp to susp queue
        if (mutex->susp == NULL){
            mutex->susp = running;
        } else {
            //I want to make sure that we add it at the end of the queue.
            green_t* current = mutex->susp;
            while(current->next != NULL){
                if (current->next == susp) {
                    break;
                }
                //printf(" current->next != %p, current = %p \n", current->next, current);
                current = current->next;
            }
            current->next = running;
            running->next = NULL;
        }

        // select the next thread for execution
        struct green_t* next = myDeQueue();

        // find the next thread
        running = next;
        swapcontext(susp->context, next->context);
    }
    // take the lock. This is enough, the caller function will wait until it gets back,
    // and do whatever with the locks.
    mutex->taken = TRUE;
    //mutex->susp = running;

    // unblock
    sigprocmask(SIG_UNBLOCK, &block, NULL);
    return 0;
}
int green_mutex_unlock(green_mutex_t *mutex) {
    // block timer interrupt
    sigprocmask(SIG_BLOCK, &block, NULL);
    green_t *susp = running;
    green_t* suspended = mutex->susp;
    // move suspended threads to ready queue
       while (suspended != NULL){
        if (suspended == susp) {
            break;
        }
        //printf(" mutex->susp != %p \n", mutex->susp);
        myEnQueue(suspended);
        if (suspended == suspended->next) {
            break;
        }
        suspended = suspended->next;
        //mutex->susp = NULL;
    }

    mutex->taken = FALSE;
    mutex->susp = NULL;
    // unblock
    sigprocmask(SIG_UNBLOCK, &block, NULL);
    return 0;
}

/*
Initialize a green condition variable.
*/
void green_cond_init(green_cond_t* cond) {
    cond = malloc(sizeof(struct green_cond_t));
}

/*
Suspend the current thread
on the condition.
*/
void green_cond_wait(green_cond_t* cond) {
    sigprocmask(SIG_BLOCK, &block, NULL);

    green_t* this = running;
    if (cond->end == NULL) {
       cond->front = cond->end = this;
       return;
    }
    cond->end->next = this;
    cond->end = this;
    //printf("this = %p \n", this);
    //run next
    struct green_t* next = myDeQueue();
    //printf("next = %p \n", next);
    running = next;
    swapcontext (this->context, next->context);
    sigprocmask(SIG_UNBLOCK, &block, NULL);
}

/*
Move the first suspended
thread to the ready cond.
*/
void green_cond_signal(green_cond_t* cond) {
    sigprocmask(SIG_BLOCK, &block, NULL);
    if (cond->front == NULL) {
       return;
    }
    struct green_t* temp = cond->front;
    cond->front = cond->front->next;
    if (cond->front == NULL) {
       cond->end = NULL;
    }
    //printf("temp = %p \n", temp);
    //printf("cond->front = %p \n", cond->front);
    myEnQueue(temp);
    sigprocmask(SIG_UNBLOCK, &block, NULL);
    //return temp;
}

/*
This function is responsible for calling the function provided by the user.

This function will do (execute) two things:
1. Start the execution of the real function (the function provided by the user).
2. After returning from the call, terminate the thread.

The tricky part is what to do when the called function returns.
The program should check if there is a thread waiting for its termination,
if so place it in the ready queue.
The stack that was allocated, and the space for the context,
should be returned to the memory management system;
the thread is now a zombie process.
There should be a thread in the ready queue
so the program select the first and schedule it for execution.
*/
void green_thread() {
    sigprocmask(SIG_UNBLOCK, &block, NULL);
    green_t* this = running;

    (*this->function)(this->arg);
    // place waiting (joining) thread in ready queue
    if (this == NULL) {
        //printf("this == NULL \n");
    }
    sigprocmask(SIG_BLOCK, &block, NULL);
    struct green_t* j = this->join;
    if (j != NULL) {
        //printf("j = %p \n", j);
        myEnQueue(j);
        //swapcontext (this->context, this->join->context);
    }
    printf ("thread %d is done \n", *(int*)this->arg);
    //printf("this = %p \n", this);
    // free alocated memory structures, i.e. free the stack.
    removeFromReadyList(this);
    free(this->context->uc_stack.ss_sp);
    free(this->context);
    // we're a zombie
    this->zombie = TRUE;
    //printf("Zombie \n");
    // find the next thread to run
    struct green_t* next = myDeQueue();
    if (next != NULL) {
        //printf("next = %p \n", next);
        running = next;
        setcontext (next->context);
    }
    //sigprocmask(SIG_UNBLOCK, &block, NULL);
    //printf("after next != NULL \n");
    //setcontext (this->context);
}

/*create a green thread*/
int green_create (green_t* new, void* (*function)(void*), void* arg) {
  //  sigprocmask(SIG_BLOCK, &block, NULL);
    ucontext_t* cntx = (ucontext_t *)malloc(sizeof(ucontext_t));
    getcontext (cntx);

    void* stack = malloc (STACK_SIZE);

    cntx->uc_stack.ss_sp = stack;
    cntx->uc_stack.ss_size = STACK_SIZE;

    makecontext (cntx, green_thread, 0);
    new->context = cntx;
    new->function = function;
    new->arg = arg;
    new->next = NULL;
    new->join = NULL;
    new->zombie = FALSE;
    // add new to the ready queue
    sigprocmask(SIG_BLOCK, &block, NULL);
    myEnQueue(new);
    sigprocmask(SIG_UNBLOCK, &block, NULL);
    return 0;
}

/*
Suspends the current thread and selects a new thread for execution.

This function will simply put the running thread last in the ready queue
and then select the first thread from the queue as the next thread to run.

The call to swapcontext() will do a context switch. It will save the
current state in susp->context and continue execution from next->context.
Note that when the suspended thread is scheduled, it will continue the execution
from exactly this point.
*/
int green_yield () {
    sigprocmask(SIG_BLOCK, &block, NULL);
    green_t* susp = running;
    // add susp to ready queue
    myEnQueue(susp);
    // select the next thread for execution
    struct green_t* next = myDeQueue();

    running = next;
    swapcontext (susp->context, next->context);
    sigprocmask(SIG_UNBLOCK, &block, NULL);
    return 0;
}

/*
The current thread is suspended waiting for a specific thread to terminate.

The join operation will wait for a thread to terminate. The program therefore add
the waiting thread to the join field to the argument and select another
thread for execution.
If the thread has already terminated we can of course continue
as if nothing happened.

The program could allow several threads to wait for the same thread.
*/
int green_join (green_t* thread) {

    if (thread->zombie) {
        return 0;
    }
    green_t* susp = running;
    sigprocmask(SIG_BLOCK, &block, NULL);
    // add to waiting threads
    //sigprocmask(SIG_BLOCK, &block, NULL);
    thread->join = susp;
    //printf ("susp = %p, susp->join = %p\n", susp, susp->join);
    // select the next thread for execution
    struct green_t* next = myDeQueue();

    running = next;
    //printf ("gonna swap\n");
    swapcontext (susp->context, next->context);
    sigprocmask(SIG_UNBLOCK, &block, NULL);
    //printf ("back in join\n");
    return 0;
}
void timer_handler(int sig) {
    sigprocmask(SIG_BLOCK, &block, NULL);
    green_t * susp = running;

    // add the running to the ready queue. Also have to
    myEnQueue(susp);

    // find the next thread for execution
    green_t* next = myDeQueue();

    running = next;
    swapcontext(susp->context, next->context);
    sigprocmask(SIG_UNBLOCK, &block, NULL);
}

/*
--------------------------------------Testing--------------------------------------
*/

//new
int flag = 0;
green_cond_t cond;
green_mutex_t mutex;
int numThreads = 10;
int counter = 0;
//end new

void* test(void* arg) {
    int id = *(int*)arg;
    int loop = 4;

    while(loop > 0) {
        if(flag == id) {
            printf("thread %d: %d \n", id, loop);
            printf("Work \n");
            loop--;
            flag = (id + 1) % 2;
            green_cond_signal(&cond);
        } else {
            //printf("thread %d: %d ", id, loop);
            //printf("Wait \n");
            green_cond_wait(&cond);
        }
    }
    //printf("while out \n");
}

void* testSharedResource(void* arg) {
  int id = *(int*)arg;
  for(int i = 0; i < 1000000; i++) {
    green_mutex_lock(&mutex);
    int temp = counter;

/*
    int dummy = 0;
    if(temp > 1337) // waste time between read and write
      dummy++;
      */
    printf(" count: %d, thread: %d, ", counter, id);
    temp++;
    counter = temp;
    green_mutex_unlock(&mutex);
  }
}

void *test_timer(void *arg){
  int i = *(int*)arg;
  int loop = MAX;
  while(loop>0){
    loop--;
    counter++;
    printf(" count: %d, thread: %d\n",counter, i);
  }
}

int main () {
    queue = malloc(sizeof *queue);
    /*green_t g0, g1;
    myEnQueue(&g0);
    printf ("queue.front %p\n", queue->end);
    myEnQueue(&g1);
    printf ("queue.front %p\n", queue->end);*/
    //end own
    green_t g0, g1;
    //green_cond_t cond;
    //printf ("cond = %p \n", cond.front);
    green_cond_init(&cond);
    green_mutex_init(&mutex);
    //printf ("cond = %p \n", cond.front);
    //green_cond_wait(&cond);
    //green_cond_signal(&cond);
    //printf ("cond = %p \n", cond.front);
    //printf ("g0 = %p, g1 = %p \n", &g0, &g1);

    int a0 = 0;
    int a1 = 1;


    green_create(&g0, testSharedResource, &a0);
    green_create(&g1, testSharedResource, &a1);

    green_join(&g0);
    green_join(&g1);
    //printf ("between\n");


    printf ("done\n");
    return 0;
}