#include "private/thd_repo.h"

	struct thd_data thd_data;

	int num_workers = 6;

	pthread_t *tids = NULL;

	thd_data.root_path = path;

	thd_data.stop = false;

	thd_data.fts = fts;

	pthread_mutex_init(&thd_data.fts_m, NULL);

	STAILQ_INIT(&thd_data.results);

	thd_data.thd_finished = 0;

	pthread_mutex_init(&thd_data.results_m, NULL);

	pthread_cond_init(&thd_data.has_result, NULL);

	/* Launch workers */

	tids = calloc(num_workers, sizeof(pthread_t));

	for (int i = 0; i < num_workers; i++) {

		pthread_create(&tids[i], NULL, (void *)&read_pkg_file, &thd_data);

	}

	for (;;) {

		struct pkg_result *r;

		pthread_mutex_lock(&thd_data.results_m);

		while ((r = STAILQ_FIRST(&thd_data.results)) == NULL) {

			if (thd_data.thd_finished == num_workers) {

				break;

			}

			pthread_cond_wait(&thd_data.has_result, &thd_data.results_m);

		}

		if (r != NULL) {

			STAILQ_REMOVE_HEAD(&thd_data.results, next);

		}

		pthread_mutex_unlock(&thd_data.results_m);

		if (r == NULL) {

			break;

		}

		if (r->retcode != EPKG_OK) {

		}

		if (run_prepared_statement(EXISTS, r->cksum) != SQLITE_ROW) {

			progress(r->pkg, data);

		pkg_get(r->pkg, PKG_ORIGIN, &origin, PKG_NAME, &name,

		    r->size, flatsize, (int64_t)licenselogic, r->cksum,

		    r->path)) != SQLITE_DONE) {

				    version, r->path)) {

		while (pkg_deps(r->pkg, &dep) == EPKG_OK) {

		while (pkg_categories(r->pkg, &category) == EPKG_OK) {

		while (pkg_licenses(r->pkg, &license) == EPKG_OK) {

		while (pkg_options(r->pkg, &option) == EPKG_OK) {

		while (pkg_shlibs(r->pkg, &shlib) == EPKG_OK) {

		pkg_free(r->pkg);

		free(r);

	// Cancel running threads

	if (retcode != EPKG_OK) {

		pthread_mutex_lock(&thd_data.fts_m);

		thd_data.stop = true;

		pthread_mutex_unlock(&thd_data.fts_m);

	}

	// Join on threads to release thread IDs

	if (tids != NULL) {

		for (int i = 0; i < num_workers; i++) {

			pthread_join(tids[i], NULL);

		}

		free(tids);

	}

void

read_pkg_file(void *data)

{

	struct thd_data *d = (struct thd_data*) data;

	struct pkg_result *r;

	FTSENT *fts_ent = NULL;

	char fts_accpath[MAXPATHLEN + 1];

	char fts_path[MAXPATHLEN + 1];

	char fts_name[MAXPATHLEN + 1];

	off_t st_size;

	int fts_info;

	struct sbuf *manifest = sbuf_new_auto();

	char *ext = NULL;

	char *pkg_path;

	for (;;) {

		fts_ent = NULL;

		/*

		 * Get a file to read from.

		 * Copy the data we need from the fts entry localy because as soon as

		 * we unlock the fts_m mutex, we can not access it.

		 */

		pthread_mutex_lock(&d->fts_m);

		if (d->stop == false) {

			fts_ent = fts_read(d->fts);

		}

		if (fts_ent != NULL) {

			strlcpy(fts_accpath, fts_ent->fts_accpath, sizeof(fts_accpath));

			strlcpy(fts_path, fts_ent->fts_path, sizeof(fts_path));

			strlcpy(fts_name, fts_ent->fts_name, sizeof(fts_name));

			st_size = fts_ent->fts_statp->st_size;

			fts_info = fts_ent->fts_info;

		}

		pthread_mutex_unlock(&d->fts_m);

		// There is no more jobs, exit the main loop.

		if (fts_ent == NULL)

				break;

		/* skip everything that is not a file */

		if (fts_info != FTS_F)

			continue;

		ext = strrchr(fts_name, '.');

		if (ext == NULL)

			continue;

		if (strcmp(ext, ".tgz") != 0 &&

				strcmp(ext, ".tbz") != 0 &&

				strcmp(ext, ".txz") != 0 &&

				strcmp(ext, ".tar") != 0)

			continue;

		if (strcmp(fts_name, "repo.txz") == 0)

			continue;

		pkg_path = fts_path;

		pkg_path += strlen(d->root_path);

		while (pkg_path[0] == '/')

			pkg_path++;

		r = calloc(1, sizeof(struct pkg_result));

		strlcpy(r->path, pkg_path, sizeof(r->path));

		r->size = st_size;

		sha256_file(fts_accpath, r->cksum);

		if (pkg_open(&r->pkg, fts_accpath, manifest) != EPKG_OK) {

			r->retcode = EPKG_WARN;

		}

		/* Add result to the FIFO and notify */

		pthread_mutex_lock(&d->results_m);

		STAILQ_INSERT_TAIL(&d->results, r, next);

		pthread_cond_signal(&d->has_result);

		pthread_mutex_unlock(&d->results_m);

	}

	/*

	 * This thread is about to exit.

	 * Notify the main thread that we are done.

	 */

	pthread_mutex_lock(&d->results_m);

	d->thd_finished++;

	pthread_cond_signal(&d->has_result);

	pthread_mutex_unlock(&d->results_m);

	sbuf_free(manifest);

}

/*

 * Copyright (c) 2012 Julien Laffaye <jlaffaye@FreeBSD.org>

 * 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

 *    in this position and unchanged.

 * 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 THE AUTHOR(S) ``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 THE AUTHOR(S) 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.

 */

#ifndef _PKG_THD_REPO_H

#define _PKG_THD_REPO_H

#include <sys/queue.h>

#include <sys/types.h>

#include <pthread.h>

struct pkg_result {

	struct pkg *pkg;

	char path[MAXPATHLEN + 1];

	char cksum[SHA256_DIGEST_LENGTH * 2 + 1];

	off_t size;

	int retcode; // to pass errors

	STAILQ_ENTRY(pkg_result) next;

};

struct thd_data {

	char *root_path;

	FTS *fts;

	bool stop;

	pthread_mutex_t fts_m; // protects `fts' and `stop'

	/* results is used as a FIFO */

	STAILQ_HEAD(results, pkg_result) results;

	int thd_finished;

	pthread_mutex_t results_m; // protects `results' an `thd_finished'

	pthread_cond_t has_result; // signal that there is at least one result

};

void read_pkg_file(void *);

#endif

		-lpthread \

		-lpthread \