bulk.c

/*
 * Copyright (c) 2019 The DragonFly Project.  All rights reserved.
 *
 * This code is derived from software contributed to The DragonFly Project
 * by Matthew Dillon <dillon@backplane.com>
 *
 * This code uses concepts and configuration based on 'synth', by
 * John R. Marino <draco@marino.st>, which was written in ada.
 *
 * 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.
 * 3. Neither the name of The DragonFly Project nor the names of its
 *    contributors may be used to endorse or promote products derived
 *    from this software without specific, prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * ``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
 * COPYRIGHT HOLDERS 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.
 */
#include "dsynth.h"

typedef struct job {
	pthread_t td;
	pthread_cond_t cond;
	bulk_t	*active;
	int terminate : 1;
} job_t;

/*
 * Most of these globals are locked with BulkMutex
 */
static int BulkScanJob;
static int BulkCurJobs;
static int BulkMaxJobs;
static job_t JobsAry[MAXBULK];
static void (*BulkFunc)(bulk_t *bulk);
static bulk_t *BulkSubmit;
static bulk_t **BulkSubmitTail = &BulkSubmit;
static bulk_t *BulkResponse;
static bulk_t **BulkResponseTail = &BulkResponse;
static pthread_cond_t BulkResponseCond;
static pthread_mutex_t BulkMutex;

static void bulkstart(void);
#if 0
static int readall(int fd, void *buf, size_t bytes);
static int writeall(int fd, const void *buf, size_t bytes);
#endif
static void *bulkthread(void *arg);

/*
 * Initialize for bulk scan operations.  Always paired with donebulk()
 */
void
initbulk(void (*func)(bulk_t *bulk), int jobs)
{
	int i;

	if (jobs > MAXBULK)
		jobs = MAXBULK;

	ddassert(BulkSubmit == NULL);
	BulkCurJobs = 0;
	BulkMaxJobs = jobs;
	BulkFunc = func;
	BulkScanJob = 0;

	addbuildenv("__MAKE_CONF", "/dev/null",
		    BENV_ENVIRONMENT | BENV_PKGLIST);

	/*
	 * CCache is a horrible unreliable hack but... leave the
	 * mechanism in-place in case someone has a death wish.
	 */
	if (UseCCache) {
		addbuildenv("WITH_CCACHE_BUILD", "yes", BENV_MAKECONF);
		addbuildenv("CCACHE_DIR", CCachePath, BENV_MAKECONF);
	}

	pthread_mutex_init(&BulkMutex, NULL);
	pthread_cond_init(&BulkResponseCond, NULL);

	pthread_mutex_lock(&BulkMutex);
	for (i = 0; i < jobs; ++i) {
		pthread_cond_init(&JobsAry[i].cond, NULL);
		pthread_create(&JobsAry[i].td, NULL, bulkthread, &JobsAry[i]);
	}
	pthread_mutex_unlock(&BulkMutex);
}

void
donebulk(void)
{
	bulk_t *bulk;
	int i;

	pthread_mutex_lock(&BulkMutex);
	while ((bulk = BulkSubmit) != NULL) {
		BulkSubmit = bulk->next;
		freebulk(bulk);
	}
	BulkSubmitTail = &BulkSubmit;

	for (i = 0; i < BulkMaxJobs; ++i) {
		JobsAry[i].terminate = 1;
		pthread_cond_signal(&JobsAry[i].cond);
	}
	pthread_mutex_unlock(&BulkMutex);
	for (i = 0; i < BulkMaxJobs; ++i) {
		pthread_join(JobsAry[i].td, NULL);
		pthread_cond_destroy(&JobsAry[i].cond);
		if (JobsAry[i].active) {
			freebulk(JobsAry[i].active);
			JobsAry[i].active = NULL;
			pthread_mutex_lock(&BulkMutex);
			--BulkCurJobs;
			pthread_mutex_unlock(&BulkMutex);
		}
		JobsAry[i].terminate = 0;
	}
	ddassert(BulkCurJobs == 0);

	while ((bulk = BulkResponse) != NULL) {
		BulkResponse = bulk->next;
		freebulk(bulk);
	}
	BulkResponseTail = &BulkResponse;

	BulkFunc = NULL;

	bzero(JobsAry, sizeof(JobsAry));

	if (UseCCache) {
		delbuildenv("WITH_CCACHE_BUILD");
		delbuildenv("CCACHE_DIR");
	}
	delbuildenv("__MAKE_CONF");
}

void
queuebulk(const char *s1, const char *s2, const char *s3, const char *s4)
{
	bulk_t *bulk;

	bulk = calloc(1, sizeof(*bulk));
	if (s1)
		bulk->s1 = strdup(s1);
	if (s2)
		bulk->s2 = strdup(s2);
	if (s3)
		bulk->s3 = strdup(s3);
	if (s4)
		bulk->s4 = strdup(s4);
	bulk->state = ONSUBMIT;

	pthread_mutex_lock(&BulkMutex);
	*BulkSubmitTail = bulk;
	BulkSubmitTail = &bulk->next;
	if (BulkCurJobs < BulkMaxJobs) {
		pthread_mutex_unlock(&BulkMutex);
		bulkstart();
	} else {
		pthread_mutex_unlock(&BulkMutex);
	}
}

/*
 * Fill any idle job slots with new jobs as available.
 */
static
void
bulkstart(void)
{
	bulk_t *bulk;
	int i;

	pthread_mutex_lock(&BulkMutex);
	while ((bulk = BulkSubmit) != NULL && BulkCurJobs < BulkMaxJobs) {
		i = BulkScanJob + 1;
		for (;;) {
			i = i % BulkMaxJobs;
			if (JobsAry[i].active == NULL)
				break;
			++i;
		}
		BulkScanJob = i;
		BulkSubmit = bulk->next;
		if (BulkSubmit == NULL)
			BulkSubmitTail = &BulkSubmit;

		bulk->state = ONRUN;
		JobsAry[i].active = bulk;
		pthread_cond_signal(&JobsAry[i].cond);
		++BulkCurJobs;
	}
	pthread_mutex_unlock(&BulkMutex);
}

/*
 * Retrieve completed job or job with activity
 */
bulk_t *
getbulk(void)
{
	bulk_t *bulk;

	pthread_mutex_lock(&BulkMutex);
	while (BulkCurJobs && BulkResponse == NULL) {
		pthread_cond_wait(&BulkResponseCond, &BulkMutex);
	}
	if (BulkResponse) {
		bulk = BulkResponse;
		ddassert(bulk->state == ONRESPONSE);
		BulkResponse = bulk->next;
		if (BulkResponse == NULL)
			BulkResponseTail = &BulkResponse;
		bulk->state = UNLISTED;
	} else {
		bulk = NULL;
	}
	pthread_mutex_unlock(&BulkMutex);
	bulkstart();

	return bulk;
}

void
freebulk(bulk_t *bulk)
{
	ddassert(bulk->state == UNLISTED);

	if (bulk->s1) {
		free(bulk->s1);
		bulk->s1 = NULL;
	}
	if (bulk->s2) {
		free(bulk->s2);
		bulk->s2 = NULL;
	}
	if (bulk->s3) {
		free(bulk->s3);
		bulk->s3 = NULL;
	}
	if (bulk->s4) {
		free(bulk->s4);
		bulk->s4 = NULL;
	}
	if (bulk->r1) {
		free(bulk->r1);
		bulk->r1 = NULL;
	}
	if (bulk->r2) {
		free(bulk->r2);
		bulk->r2 = NULL;
	}
	if (bulk->r3) {
		free(bulk->r3);
		bulk->r3 = NULL;
	}
	if (bulk->r4) {
		free(bulk->r4);
		bulk->r4 = NULL;
	}
	free(bulk);
}

#if 0

/*
 * Returns non-zero if unable to read specified number of bytes
 */
static
int
readall(int fd, void *buf, size_t bytes)
{
	ssize_t r;

	for (;;) {
		r = read(fd, buf, bytes);
		if (r == (ssize_t)bytes)
			break;
		if (r > 0) {
			buf = (char *)buf + r;
			bytes -= r;
			continue;
		}
		if (r < 0 && errno == EINTR)
			continue;
		return 1;
	}
	return 0;
}

static
int
writeall(int fd, const void *buf, size_t bytes)
{
	ssize_t r;

	for (;;) {
		r = write(fd, buf, bytes);
		if (r == (ssize_t)bytes)
			break;
		if (r > 0) {
			buf = (const char *)buf + r;
			bytes -= r;
			continue;
		}
		if (r < 0 && errno == EINTR)
			continue;
		return 1;
	}
	return 0;
}

#endif

static void *
bulkthread(void *arg)
{
	job_t *job = arg;
	bulk_t *bulk;

	pthread_mutex_lock(&BulkMutex);
	for (;;) {
		if (job->terminate)
			break;
		if (job->active == NULL)
			pthread_cond_wait(&job->cond, &BulkMutex);
		bulk = job->active;
		if (bulk) {
			bulk->state = ISRUNNING;

			pthread_mutex_unlock(&BulkMutex);
			BulkFunc(bulk);
			pthread_mutex_lock(&BulkMutex);

			bulk->state = ONRESPONSE;
			bulk->next = NULL;
			*BulkResponseTail = bulk;
			BulkResponseTail = &bulk->next;
			--BulkCurJobs;
			pthread_cond_signal(&BulkResponseCond);
		}

		/*
		 * Optimization - automatically fetch the next job
		 */
		if ((bulk = BulkSubmit) != NULL && job->terminate == 0) {
			BulkSubmit = bulk->next;
			if (BulkSubmit == NULL)
				BulkSubmitTail = &BulkSubmit;
			bulk->state = ONRUN;
			job->active = bulk;
			++BulkCurJobs;
		} else {
			job->active = NULL;
		}
	}
	pthread_mutex_unlock(&BulkMutex);

	return NULL;
}