Archipelago

xseg-objs := xsegmod.o xq.k.o xpool.k.o xhash.k.o xheap.k.o xobj.k.o xseg.k.o xworkq.k.o xwaitq.k.o 

default: xq.k.c xpool.k.c xhash.k.c xheap.k.c xobj.k.c xworkq.k.c xwaitq.k.c xseg.k.c

xwaitq.k.c: $(BASE)/xtypes/xwaitq.c $(BASE)/xtypes/xwaitq.h

	ln -sf $< $@

	rm -f xq.k.c xpool.k.c xhash.k.c xheap.k.c xobj.k.c xworkq.k.c xwaitq.k.c xseg.k.c

	make -C xwaitq all

xwaitq/xwaitq.o:

	make -C xwaitq xwaitq.o

xwaitq/xwaitq.pic.o:

	make -C xwaitq xwaitq.pic.o

		$(BASE)/xtypes/xobj_exports.h $(BASE)/xtypes/xworkq_exports.h\

		$(BASE)/xtypes/xwaitq_exports.h

		$(BASE)/xtypes/xwaitq_exports.h | ./make_symbol_map.sh > $@

				 xworkq/xworkq.pic.o xwaitq/xwaitq.pic.o $(DRVOBJS)

			xworkq/xworkq.pic.o xwaitq/xwaitq.pic.o \

	make -C xwaitq install-src

	make -C xwaitq clean

# Copyright 2012 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

#

.PHONY: default all clean install install-src

include $(XSEG_HOME)/base.mk

DEBUG=-g

FILES="Makefile"

#FILES+=$(shell ls *.h)

#FILES+=$(shell ls *.c)

SUBDIR:=$(subst $(XSEG_HOME),,$(CURDIR))

default: all

all: xwaitq.o xwaitq.pic.o xwaitq_test

$(BASE)/sys/user/xseg_user.o:

	make -C $(BASE)/sys/user xseg_user.o

xwaitq_test: $(BASE)/xtypes/xwaitq_test.c xwaitq.o $(BASE)/sys/user/xq/xq.o $(BASE)/sys/user/xseg_user.o

	$(CC) $(CFLAGS) $(INC) -L$(LIB) -o $@ $< xwaitq.o \

	$(BASE)/sys/user/xseg_user.o $(BASE)/sys/user/xq/xq.o \

	-ldl -lpthread

xwaitq.o: $(BASE)/xtypes/xwaitq.c $(BASE)/xtypes/xwaitq.h  $(BASE)/xtypes/xwork.h $(BASE)/xtypes/xlock.h

	$(CC) $(CFLAGS) $(INC) -c -o $@ $<

xwaitq.pic.o: $(BASE)/xtypes/xwaitq.c $(BASE)/xtypes/xwaitq.h $(BASE)/xtypes/xwork.h $(BASE)/xtypes/xlock.h

	$(CC) $(CFLAGS) $(INC) -fPIC -c -o $@ $<

clean:

	rm -f xwaitq.o xwaitq.pic.o xwaitq_test

install:

install-src:

	install -d $(DESTDIR)$(srcdir)$(SUBDIR) ;

	@for f in $(FILES) ; do \

		install -o 0 -g 0 -m 644 -t $(DESTDIR)$(srcdir)$(SUBDIR) $$f ; \

	done

#include <xtypes/domain.h>

#include <xtypes/xwaitq.h>

static int __check_cond(struct xwaitq *wq)

{

	return wq->cond_fn(wq->cond_arg);

}

int xwaitq_init(struct xwaitq *wq, int (*cond_fn)(void *arg), void *arg, uint32_t flags)

{

	wq->cond_fn = cond_fn;

	wq->cond_arg = arg;

	wq->flags = flags;

	wq->q = xtypes_malloc(sizeof(struct xq));

	if (!wq->q)

		return -1;

	xlock_release(&wq->lock);

	if (!xq_alloc_empty(wq->q, 8)){

		xtypes_free(wq->q);

		return -1;

	}

	return 0;

}

void xwaitq_destroy(struct xwaitq *wq)

{

	xq_free(wq->q);

	xtypes_free(wq->q);

}

int __xwaitq_enqueue(struct xwaitq *wq, struct work *w)

{

	//enqueure and resize if necessary

	xqindex r;

	struct xq *newq;

	r = __xq_append_tail(wq->q, (xqindex)w);

	if (r == Noneidx){

		newq = xtypes_malloc(sizeof(struct xq));

		if (!newq){

			return -1;

		}

		if (!xq_alloc_empty(newq, wq->q->size*2)){

			xtypes_free(newq);

			return -1;

		}

		if (__xq_resize(wq->q, newq) == Noneidx){

			xq_free(newq);

			xtypes_free(newq);

			return -1;

		}

		xtypes_free(wq->q);

		wq->q = newq;

		r = __xq_append_tail(wq->q, (xqindex)w);

	}

	return ((r == Noneidx)? -1 : 0);

}

int xwaitq_enqueue(struct xwaitq *wq, struct work *w)

{

	int r;

	xlock_acquire(&wq->lock, 1);

	if (__check_cond(wq)){

		xlock_release(&wq->lock);

		w->job_fn(w->job);

		return 0;

	}

	r = __xwaitq_enqueue(wq, w);

	xlock_release(&wq->lock);

	return r;

}

void xwaitq_signal(struct xwaitq *wq)

{

	xqindex xqi;

	struct work *w;

	if (!xq_count(wq->q))

		return;

	if (wq->flags & XWAIT_SIGNAL_ONE){

		if (!xlock_try_lock(&wq->lock, 1))

			return;

	} else {

		xlock_acquire(&wq->lock, 1);

	}

	while (xq_count(wq->q) && __check_cond(wq)){

		xqi = __xq_pop_head(wq->q);

		xlock_release(&wq->lock);

		if (xqi == Noneidx){

			break;

		}

		w = (struct work *)xqi;

		w->job_fn(w->job);

		if (wq->flags & XWAIT_SIGNAL_ONE){

			if (!xlock_try_lock(&wq->lock, 1))

				return;

		} else {

			xlock_acquire(&wq->lock, 1);

		}

	}

	xlock_release(&wq->lock);

}

#ifndef __X_WAITQ_H

#define __X_WAITQ_H

#include <xtypes/xq.h>

#include <xtypes/xwork.h>

#define XWAIT_SIGNAL_ONE (1 << 0)

struct xwaitq {

	int (*cond_fn)(void *arg);

	void *cond_arg;

	uint32_t flags;

	struct xq *q;

	struct xlock lock;

};

int xwaitq_init(struct xwaitq *wq, int (*cond_fn)(void *arg), void *arg, uint32_t flags);

int xwaitq_enqueue(struct xwaitq *wq, struct work *w);

void xwaitq_signal(struct xwaitq *wq);

void xwaitq_destroy(struct xwaitq *wq);

#endif /* __X_WAITQ_H */

/*

 * Copyright 2013 GRNET S.A. All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or

 * without modification, are permitted provided that the following

 * conditions are met:

 *

 *   1. Redistributions of source code must retain the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer.

 *   2. Redistributions in binary form must reproduce the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer in the documentation and/or other materials

 *      provided with the distribution.

 *

 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 * POSSIBILITY OF SUCH DAMAGE.

 *

 * The views and conclusions contained in the software and

 * documentation are those of the authors and should not be

 * interpreted as representing official policies, either expressed

 * or implied, of GRNET S.A.

 */

EXPORT_SYMBOL(xwaitq_init);

EXPORT_SYMBOL(xwaitq_destroy);

EXPORT_SYMBOL(xwaitq_enqueue);

EXPORT_SYMBOL(xwaitq_signal);

#include "xwaitq.h"

#include <stdio.h>

#include <stdlib.h>

#include <pthread.h>

#include <xtypes/xlock.h>

#include <sys/time.h>

volatile int cond = 0;

unsigned long sum = 0;

struct xlock lock;

int condfn(void *arg)

{

	return cond;

}

void jobfn(void *arg)

{

	unsigned long c = (unsigned long) arg;

	xlock_acquire(&lock, c);

	sum += c;

	xlock_release(&lock);

}

int test1(unsigned long n)

{

	struct xwaitq wq;

	unsigned long i;

	struct work *works = malloc(sizeof(struct work) * n);

	xwaitq_init(&wq, condfn, NULL, 0);

	cond = 0;

	sum = 0;

	xlock_release(&lock);

	for (i = 0; i < n; i++) {

		works[i].job_fn = jobfn;

		works[i].job = 1;

	}

	for (i = 0; i < n; i++) {

		xwaitq_enqueue(&wq, &works[i]);

	}

	cond = 1;

	xwaitq_signal(&wq);

	free(works);

	xwaitq_destroy(&wq);

	return ((sum == n)? 0 : -1);

}

struct thread_arg{

	struct xwaitq *wq;

	unsigned long n;

	unsigned long num;

};

void *thread_test(void *arg)

{

	struct thread_arg *targ = (struct thread_arg *)arg;

	unsigned long n = targ->n;

	unsigned long i;

	struct work *works = malloc(sizeof(struct work) * n);

	for (i = 0; i < n; i++) {

		works[i].job_fn = jobfn;

		works[i].job = targ->num;

	}

	for (i = 0; i < n; i++) {

		xwaitq_enqueue(targ->wq, &works[i]);

	}

	cond = 1;

	xwaitq_signal(targ->wq);

//	free(works);

	return NULL;

}

int test2(unsigned long n, unsigned long nr_threads)

{

	int i, r;

	struct xwaitq wq;

	xwaitq_init(&wq, condfn, NULL, 0);

	cond = 0;

	sum = 0;

	xlock_release(&lock);

	struct thread_arg *targs = malloc(sizeof(struct thread_arg)*nr_threads * n);

	pthread_t *threads = malloc(sizeof(pthread_t) * nr_threads);

	for (i = 0; i < nr_threads; i++) {

		targs[i].num = i+1;

		targs[i].n = n;

		targs[i].wq = &wq;

	}

	for (i = 0; i < nr_threads; i++) {

		r = pthread_create(&threads[i], NULL, thread_test, &targs[i]);

		if (r) {

			fprintf(stderr, "error pthread_create\n");

			return -1;

		}

	}

	for (i = 0; i < nr_threads; i++) {

		pthread_join(threads[i], NULL);

	}

	free(targs);

	free(threads);

	xwaitq_destroy(&wq);

	unsigned long expected_sum = 0;

	for (i = 0; i < nr_threads; i++) {

		expected_sum += n*(i+1);

	}

	return ((sum == expected_sum) ? 0 : -1);

}

int test3(unsigned long n, unsigned long nr_threads)

{

	int i, r;

	struct xwaitq wq;

	xwaitq_init(&wq, condfn, NULL, XWAIT_SIGNAL_ONE);

	cond = 0;

	sum = 0;

	xlock_release(&lock);

	struct thread_arg *targs = malloc(sizeof(struct thread_arg)*nr_threads * n);

	pthread_t *threads = malloc(sizeof(pthread_t) * nr_threads);

	for (i = 0; i < nr_threads; i++) {

		targs[i].num = i+1;

		targs[i].n = n;

		targs[i].wq = &wq;

	}

	for (i = 0; i < nr_threads; i++) {

		r = pthread_create(&threads[i], NULL, thread_test, &targs[i]);

		if (r) {

			fprintf(stderr, "error pthread_create\n");

			return -1;

		}

	}

	for (i = 0; i < nr_threads; i++) {

		pthread_join(threads[i], NULL);

	}

	free(targs);

	free(threads);

	xwaitq_destroy(&wq);

	unsigned long expected_sum = 0;

	for (i = 0; i < nr_threads; i++) {

		expected_sum += n*(i+1);

	}

	return ((sum == expected_sum) ? 0 : -1);

}

int main(int argc, const char *argv[])

{

	struct timeval start, end, tv;

	int r;

	int n = atoi(argv[1]);

	int t = atoi(argv[2]);

	fprintf(stderr, "Running test1\n");

	gettimeofday(&start, NULL);

	r = test1(n);

	if (r < 0){

		fprintf(stderr, "Test1: FAILED\n");

		return -1;

	}

	gettimeofday(&end, NULL);

	timersub(&end, &start, &tv);

	fprintf(stderr, "Test1: PASSED\n");

	fprintf(stderr, "Test time: %ds %dusec\n\n", (int)tv.tv_sec, (int)tv.tv_usec);

	fprintf(stderr, "running test2\n");

	gettimeofday(&start, NULL);

	r = test2(n, t);

	if (r < 0){

		fprintf(stderr, "test2: failed\n");

		return -1;

	}

	gettimeofday(&end, NULL);

	fprintf(stderr, "test2: passed\n");

	timersub(&end, &start, &tv);

	fprintf(stderr, "Test time: %ds %dusec\n\n", (int)tv.tv_sec, (int)tv.tv_usec);

	fprintf(stderr, "running test3\n");

	gettimeofday(&start, NULL);

	r = test3(n, t);

	if (r < 0){

		fprintf(stderr, "test3: failed\n");

		return -1;

	}

	gettimeofday(&end, NULL);

	fprintf(stderr, "test3: passed\n");

	timersub(&end, &start, &tv);

	fprintf(stderr, "Test time: %ds %dusec\n\n", (int)tv.tv_sec, (int)tv.tv_usec);

	return 0;

}