#undef __STRICT_ANSI__

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "chan.h"


int passed = 0;

void assert_true(int expression, chan_t* chan, char* msg)
{
    if (!expression)
    {
        chan_dispose(chan);
        fprintf(stderr, "Assertion failed: %s\n", msg);
        exit(1);
    }
}

void pass()
{
    printf(".");
    fflush(stdout);
    passed++;
}

void wait_for_reader(chan_t* chan)
{
    for (;;)
    {
        pthread_mutex_lock(&chan->m_mu);
        int send = chan->r_waiting > 0;
        pthread_mutex_unlock(&chan->m_mu);
        if (send) break;
        sched_yield();
    }
}

void wait_for_writer(chan_t* chan)
{
    for (;;)
    {
        pthread_mutex_lock(&chan->m_mu);
        int recv = chan->w_waiting > 0;
        pthread_mutex_unlock(&chan->m_mu);
        if (recv) break;
        sched_yield();
    }
}

void test_chan_init_buffered()
{
    size_t size = 5;
    chan_t* chan = chan_init(size);

    assert_true(chan->queue != NULL, chan, "Queue is NULL");
    assert_true((size_t) chan->queue->capacity == size, chan, "Size is not 5");
    assert_true(!chan->closed, chan, "Chan is closed");

    chan_dispose(chan);
    pass();
}

void test_chan_init_unbuffered()
{
    chan_t* chan = chan_init(0);

    assert_true(chan->queue == NULL, chan, "Queue is not NULL");
    assert_true(!chan->closed, chan, "Chan is closed");

    chan_dispose(chan);
    pass();
}

void test_chan_init()
{
    test_chan_init_buffered();
    test_chan_init_unbuffered();
}

void test_chan_close()
{
    chan_t* chan = chan_init(0);

    assert_true(!chan->closed, chan, "Chan is closed");
    assert_true(!chan_is_closed(chan), chan, "Chan is closed");
    assert_true(chan_close(chan) == 0, chan, "Close failed");
    assert_true(chan->closed, chan, "Chan is not closed");
    assert_true(chan_close(chan) == -1, chan, "Close succeeded");
    assert_true(chan_is_closed(chan), chan, "Chan is not closed");
    
    chan_dispose(chan);
    pass();
}

void test_chan_send_buffered()
{
    chan_t* chan = chan_init(1);
    void* msg = "foo";

    assert_true(chan_size(chan) == 0, chan, "Queue is not empty");
    assert_true(chan_send(chan, msg) == 0, chan, "Send failed");
    assert_true(chan_size(chan) == 1, chan, "Queue is empty");
    
    chan_dispose(chan);
    pass();
}

void* receiver(void* chan)
{
    void* msg;
    chan_recv(chan, &msg);
    return NULL;
}

void test_chan_send_unbuffered()
{
    chan_t* chan = chan_init(0);
    void* msg = "foo";
    pthread_t th;
    pthread_create(&th, NULL, receiver, chan);

    wait_for_reader(chan);

    assert_true(chan_size(chan) == 0, chan, "Chan size is not 0");
    assert_true(!chan->w_waiting, chan, "Chan has sender");
    assert_true(chan_send(chan, msg) == 0, chan, "Send failed");
    assert_true(!chan->w_waiting, chan, "Chan has sender");
    assert_true(chan_size(chan) == 0, chan, "Chan size is not 0");

    pthread_join(th, NULL);
    chan_dispose(chan);
    pass();
}

void test_chan_send()
{
    test_chan_send_buffered();
    test_chan_send_unbuffered();
}

void test_chan_recv_buffered()
{
    chan_t* chan = chan_init(1);
    void* msg = "foo";

    assert_true(chan_size(chan) == 0, chan, "Queue is not empty");
    chan_send(chan, msg);
    assert_true(chan_size(chan) == 1, chan, "Queue is empty");
    void* received;
    assert_true(chan_recv(chan, &received) == 0, chan, "Recv failed");
    assert_true(msg == received, chan, "Messages are not equal");
    assert_true(chan_size(chan) == 0, chan, "Queue is not empty");
    
    chan_dispose(chan);
    pass();
}

void* sender(void* chan)
{
    chan_send(chan, "foo");
    return NULL;
}

void test_chan_recv_unbuffered()
{
    chan_t* chan = chan_init(0);
    pthread_t th;
    pthread_create(&th, NULL, sender, chan);

    assert_true(chan_size(chan) == 0, chan, "Chan size is not 0");
    assert_true(!chan->r_waiting, chan, "Chan has reader");
    void *msg;
    assert_true(chan_recv(chan, &msg) == 0, chan, "Recv failed");
    assert_true(strcmp(msg, "foo") == 0, chan, "Messages are not equal");
    assert_true(!chan->r_waiting, chan, "Chan has reader");
    assert_true(chan_size(chan) == 0, chan, "Chan size is not 0");

    pthread_join(th, NULL);
    chan_dispose(chan);
    pass();
}

void test_chan_recv()
{
    test_chan_recv_buffered();
    test_chan_recv_unbuffered();
}

void test_chan_select_recv()
{
    chan_t* chan1 = chan_init(0);
    chan_t* chan2 = chan_init(1);
    chan_t* chans[2] = {chan1, chan2};
    void* recv;
    void* msg = "foo";

    chan_send(chan2, msg);

    switch(chan_select(chans, 2, &recv, NULL, 0, NULL))
    {
        case 0:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Received on wrong channel\n");
            exit(1);
        case 1:
            if (strcmp(recv, msg) != 0)
            {
                chan_dispose(chan1);
                chan_dispose(chan2);
                fprintf(stderr, "Messages are not equal\n");
                exit(1);
            }
            recv = NULL;
            break;
        default:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Received on no channels\n");
            exit(1);
    }

    pthread_t th;
    pthread_create(&th, NULL, sender, chan1);
    wait_for_writer(chan1);

    switch(chan_select(chans, 2, &recv, NULL, 0, NULL))
    {
        case 0:
            if (strcmp(recv, "foo") != 0)
            {
                chan_dispose(chan1);
                chan_dispose(chan2);
                fprintf(stderr, "Messages are not equal\n");
                exit(1);
            }
            recv = NULL;
            break;
        case 1:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Received on wrong channel\n");
            exit(1);
        default:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Received on no channels\n");
            exit(1);
    }

    switch(chan_select(&chan2, 1, &msg, NULL, 0, NULL))
    {
        case 0:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Received on channel\n");
            exit(1);
        default:
            break;
    }

    pthread_join(th, NULL);
    chan_dispose(chan1);
    chan_dispose(chan2);
    pass();
}

void test_chan_select_send()
{
    chan_t* chan1 = chan_init(0);
    chan_t* chan2 = chan_init(1);
    chan_t* chans[2] = {chan1, chan2};
    void* msg[] = {"foo", "bar"};

    switch(chan_select(NULL, 0, NULL, chans, 2, msg))
    {
        case 0:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Sent on wrong channel");
            exit(1);
        case 1:
            break;
        default:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Sent on no channels");
            exit(1);
    }

    void* recv;
    chan_recv(chan2, &recv);
    if (strcmp(recv, "bar") != 0)
    {
        chan_dispose(chan1);
        chan_dispose(chan2);
        fprintf(stderr, "Messages are not equal");
        exit(1);
    }

    chan_send(chan2, "foo");

    pthread_t th;
    pthread_create(&th, NULL, receiver, chan1);
    wait_for_reader(chan1);

    switch(chan_select(NULL, 0, NULL, chans, 2, msg))
    {
        case 0:
            break;
        case 1:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Sent on wrong channel");
            exit(1);
        default:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Sent on no channels");
            exit(1);
    }

    switch(chan_select(NULL, 0, NULL, &chan1, 1, msg[0]))
    {
        case 0:
            chan_dispose(chan1);
            chan_dispose(chan2);
            fprintf(stderr, "Sent on channel");
            exit(1);
        default:
            break;
    }

    pthread_join(th, NULL);
    chan_dispose(chan1);
    chan_dispose(chan2);
    pass();
}

void test_chan_select()
{
    test_chan_select_recv();
    test_chan_select_send();
}

void test_chan_int()
{
    chan_t* chan = chan_init(1);
    int s = 12345, r = 0;
    chan_send_int(chan, s);
    chan_recv_int(chan, &r);
    assert_true(s == r, chan, "Wrong value of int(12345)");

    int32_t s32 = 12345, r32 = 0;
    chan_send_int32(chan, s32);
    chan_recv_int32(chan, &r32);
    assert_true(s32 == r32, chan, "Wrong value of int32(12345)");

    int64_t s64 = 12345, r64 = 0;
    chan_send_int64(chan, s64);
    chan_recv_int64(chan, &r64);
    assert_true(s64 == r64, chan, "Wrong value of int64(12345)");

    chan_dispose(chan);
    pass();
}

void test_chan_double()
{
    chan_t* chan = chan_init(1);
    double s = 123.45, r = 0;
    chan_send_double(chan, s);
    chan_recv_double(chan, &r);
    assert_true(s == r, chan, "Wrong value of double(123.45)");

    chan_dispose(chan);
    pass();
}

void test_chan_buf()
{
    chan_t* chan = chan_init(1);
    char s[256], r[256];
    strcpy(s, "hello world");
    chan_send_buf(chan, s, sizeof(s));
    strcpy(s, "Hello World");
    chan_recv_buf(chan, &r, sizeof(s));
    assert_true(memcmp(s, r, sizeof(s)), chan, "Wrong value of buf");

    chan_dispose(chan);
    pass();
}

void test_chan_multi()
{
    chan_t* chan = chan_init(5);
    pthread_t th[100];
    for (int i = 0; i < 50; ++i)
    {
       pthread_create(&th[i], NULL, sender, chan);
    }

    for (;;)
    {
       pthread_mutex_lock(&chan->m_mu);
       int all_waiting = chan->w_waiting == 45;
       pthread_mutex_unlock(&chan->m_mu);
       if (all_waiting) break;
       sched_yield();
    }

    for (int i = 50; i < 100; ++i)
    {
       pthread_create(&th[i], NULL, receiver, chan);
    }

    for (int i = 0; i < 100; ++i)
    {
       pthread_join(th[i], NULL);
    }

    chan_dispose(chan);
    pass();
}

void test_chan_multi2()
{
    chan_t* chan = chan_init(5);
    pthread_t th[100];
    for (int i = 0; i < 100; ++i)
    {
        pthread_create(&th[i], NULL, receiver, chan);
    }

    for (;;)
    {
       pthread_mutex_lock(&chan->m_mu);
       int all_waiting = chan->r_waiting == 100;
       pthread_mutex_unlock(&chan->m_mu);
       if (all_waiting) break;
       sched_yield();
    }

    // Simulate 5 high-priority writing threads.
    pthread_mutex_lock(&chan->m_mu);
    for (int i = 0; i < 5; ++i)
    {
        assert_true(0 == queue_add(chan->queue, "foo"), chan,
            "Simulate writer thread");
        pthread_cond_signal(&chan->r_cond); // wakeup reader
    }

    // Simulate 6th high-priority waiting writer.
    assert_true(chan->queue->size == chan->queue->capacity, chan,
        "6th writer has to wait");
    chan->w_waiting++;
    // Simulated writer must be woken up by reader.
    pthread_cond_wait(&chan->w_cond, &chan->m_mu);
    chan->w_waiting--;
    pthread_mutex_unlock(&chan->m_mu);

    // Wake up other waiting reader.
    for (int i = 5; i < 100; ++i)
    {
        chan_send(chan, "foo");
    }

    for (int i = 0; i < 100; ++i)
    {
        pthread_join(th[i], NULL);
    }

    chan_dispose(chan);
    pass();
}

int main()
{
    test_chan_init();
    test_chan_close();
    test_chan_send();
    test_chan_recv();
    test_chan_select();
    test_chan_int();
    test_chan_double();
    test_chan_buf();
    test_chan_multi();
    test_chan_multi2();
    printf("\n%d passed\n", passed);
    return 0;
}