thq.c revision 7c478bd95313f5f23a4c958a745db2134aa03244
/*
* CDDL HEADER START
*
* The contents of this file are subject to the terms of the
* Common Development and Distribution License, Version 1.0 only
* (the "License"). You may not use this file except in compliance
* with the License.
*
* You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
* See the License for the specific language governing permissions
* and limitations under the License.
*
* When distributing Covered Code, include this CDDL HEADER in each
* file and include the License file at usr/src/OPENSOLARIS.LICENSE.
* If applicable, add the following below this CDDL HEADER, with the
* fields enclosed by brackets "[]" replaced with your own identifying
* information: Portions Copyright [yyyy] [name of copyright owner]
*
* CDDL HEADER END
*/
/*
* Copyright 1999-2002 Sun Microsystems, Inc.
* All rights reserved.
* Use is subject to license terms.
*/
#pragma ident "%Z%%M% %I% %E% SMI"
/*
* file: thq.c
*
* This file contains all of the routines used for thread
* queue management. Thread queue management provides an
* interface for an application to dispatch processing of a
* data object to a pool of threads.
*/
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <signal.h>
#include <pthread.h>
#include <unistd.h>
#include "thq.h"
#define DEFAULT_MIN_THREADS 10
#define DEFAULT_MAX_THREADS 100
/*
* Function: tq_alloc
*
* Arguments: action - Pointer to the function that new threads
* must call.
* end - Pointer to a function that will be called
* if threads are being killed (optional).
* arg - Pointer that will be passed as an argument
* to new threads created (optional).
* shutdown - Pointer to the shutdown flag. If this
* is provided, if the pointer contains a
* non-zero value, the thread queueing system
* will assume that the process is shutting
* down (optional).
* max - The maximum number of threads
* min - The minimum number of threads
*
* Description: This function is used to create the thread
* queue. The end, arg and shutdown parameters
* MAY be set to NULL. If the max and min are
* not provided, we will use our own defaults.
*
* Once the thread queue has been allocated, we
* will initialize it, and init the the lock and
* condition variables.
*
* Returns: Upon successful completion, the function will
* return a pointer to a thread queue, otherwise
* NULL
*/
{
/* no function passed, error */
return (NULL);
} /* end if */
/* if those parameters are not passed, take default ones */
return (NULL);
} /* end if */
return (NULL);
} /* end if */
}
/*
* The wd_mask mechanism is used to accelerate the search of
* a "hole" in the queue->tid[] structure whenever we needed
* to create a new thread and to accelerate the search of a
* tid in the queue->tid[] structure whenever we needed to
* end a thread. The previous code(i.e., wd_mask = 0) was
* always parsing this structure sequentially with the
* consequent problems:
*
* + in thr creations, "holes" where not uniformly distributed
* through the queue->tid[] structure BUT condensed at the end
* of it.
*
* + in thr_ ends, finding the thread tid associated to the
* thread to be ended was not optimal
*
* With the wd_mask mechanism, on the contrary:
*
* + in thr creations, the new assigned tid is itself used
* for the indexation on the queue->tid[] and stored at the
* first "hole" found from that position. If we get to
* queue->tid[MAX], we round in circle and come back to
* queue->tid[0]
*
* + in thr_deletions, the old pid is itself used for the
* indexation on the queue->tid[], accelerating the search.
*
* This mechanism is much faster than the old one, specially
* considering that, as tid assignments are(were! when not
* detached) sequential, if the queue->tid[] contains 256
* elements, by the time the thread with tid 259 will be
* created there are enormous chances that the thread(259-256
* = 3) with tid 3 has already ended, hence, finding a
* "hole" in the queue->tid[] structure.
*/
sizeof (pthread_t));
sizeof (pthread_t));
sizeof (pthread_t));
sizeof (pthread_t));
sizeof (pthread_t));
sizeof (pthread_t));
sizeof (pthread_t));
} else if (max < 4096) {
sizeof (pthread_t));
} else if (max < 8192) {
sizeof (pthread_t));
} else if (max < 16384) {
sizeof (pthread_t));
} else {
sizeof (pthread_t));
}
} /* end if */
return (NULL);
} /* end if */
} /* end if */
return (queue);
}
/*
* Function: dump_queue
*
* Arguments: queue - Pointer to the thread queue
*
* Description: This function is provided for debugging purposes
* but is not currently used in the code, and will
* print out the contents of the thread queue.
*
* Returns:
*/
#ifndef lint
void
{
} /* end for */
(void) printf("\n");
}
#endif
/*
* Function: timedthread
*
* Arguments: arg - Pointer to an argument that will be
* provided to the new thread.
*
* Description: This strange function will sleep for 4 seconds
* before calling the thread callback routine.
*
* Returns: should never return.
*/
static void
timedthread(void * arg)
{
(void) sleep(4);
}
/*
* Function: create_new_thread
*
* Arguments: queue - Pointer to the thread queue
* timer - Determines whether we need to
* wait before we call the new
* thread.
*
* Description: This function is used to create a new
* thread. If the number of threads waiting
* is smaller than the number of items in
* our queue, and we have not reached our
* maximum number of threads, a new thread
* will be created.
*
* Returns: int, 0 if successful
*/
static int
{
int i;
/* create a thread to manage this request */
if (timer) {
(void *(*)())timedthread, queue) != 0) {
return (-1);
} /* end if */
} /* end if */
return (-1);
} /* end if */
if (pthread_detach(tid)) {
/*
* Unable to detach the thread, let's kill it.
*/
(void) pthread_cancel(tid);
}
/*
* put the thread id into the saved array for the
* watchdog mechanism
*/
break;
} /* end if */
} /* end for */
for (i = 0; i < htid; ++i) {
break;
} /* end if */
} /* end for */
}
} else {
break;
} /* end if */
} /* end for */
}
} /* end if */
return (0);
}
/*
* Function: tq_queue
*
* Arguments: queue - Pointer to the thread queue
* arg - Pointer to object to be processed
*
* Description: queue an action to be done in the arg queue
* and signal any waiting thread that something
* is to be processed. If we notice that we've
* reached our maximum number of threads, and no
* threads are in the waiting state, we will
* make sure that all threads stil exist.
*
* if arg is null, special case, just send the signal
* (probably in case of shutdown)
*
* Returns: int, 0 if successful
*/
int
{
int i;
static int times_called = 0;
static int gc = 0;
if (!gc) {
times_called = 0;
gc = 1;
}
}
return (0);
} /* end if */
/*
* shutdown condition, broadcast to all the threads waiting to
* terminate them
* and refuse new stuff in the queue
*/
return (-1);
}
if (arg) {
return (-1);
} /* end if */
} /* end if */
} /* end if */
++(queue->queue_size);
if (queue->thr_waiting == 0 &&
/*
* Since it is possible that some threads have exited
* by themselves, if we reach the maximum number of
* threads, we will kill with signal 0, which simply
* ensures that the thread is still valid. If the
* thread is no longer valid, we will free the entry
* in the queue.
*/
int max;
gc = 0;
for (i = 0; i < max; ++i) {
} /* end if */
} /* end if */
} /* end for */
} /* end if */
/*
* if size of the queue > nb of thread waiting and we
* didn't reach the maximum number of threads, create a
* new one
*/
if (create_new_thread(queue, 0)) {
return (-1);
}
} /* end if */
else
/* just signal one of the waiting threads */
return (0);
}
/*
* Function: tq_end_thread
*
* Arguments: queue - Pointer to the thread queue
* endit_arg - parameter to be passed to
* the thread shutdown function.
*
* Description: This function is called when a thread
* needs to be shutdown. if a termination
* function was provided during tq_alloc(),
* we will call it with the argument provided.
*
* If the shutdown flag is not set, we will
* create another thread, otherwise we will
* send a broadcast that we are shutting down.
*
* Returns:
*/
static void
{
int i;
/*
* call the finish function of the thread
*/
} /* end if */
break;
} /* end if */
} /* end for */
for (i = 0; i < htid; ++i) {
break;
} /* end if */
} /* end for */
}
} else {
break;
} /* end if */
} /* end for */
}
/*
* It is possible that while we started to shutdown, a new item
* showed up on the queue. By calling create_new_thread() we
* will create such a thread, only if necessary.
*/
/*
* and terminate the thread
*/
/* shutdown condition, warn the waiting function if any */
}
/*
* update number of active threads
*/
--(queue->thr_waiting);
/*
* unlock the queue
*/
pthread_exit(0);
} /* end static void tq_end_thread */
/*
* Function: tq_dequeue
*
* Arguments: queue - Pointer to the thread queue
* endit_arg - parameter to be passed to
* the thread shutdown function.
*
* Description: This function is called by the threads
* to retrieve an object to process. If
* the caller ends up waiting for 30 seconds
* without processing anything, we will
* terminate the thread.
*
* Returns: a pointer to an object to process taken
* from the queue.
*/
void *
{
void * arg;
return (NULL);
} /* end if */
++(queue->thr_waiting);
/* dump_queue(queue); */
/*
* if something in the queue, dequeue it and return the
* action to be done
*/
(void) pthread_cond_broadcast(
} /* end if */
} /* end if */
--(queue->queue_size);
--(queue->thr_waiting);
return (arg);
} else {
int rc;
/*
* wait for the condition arg_loop to be set
* or a signal occurs, or a timeout
*/
/*
* we don't want to keep this thread alive
* call shutdown function, release the lock
* and exit
*/
}
} /* end if */
} /* end while */
/*
* shutdown condition
*
* end the thread
* call shutdown function, release the lock and exit
*/
return (NULL); /* never called, just here for the compiler */
}
/*
* Function: tq_shutdown
*
* Arguments: queue - Pointer to the thread queue
* immediate - shutdown immediately flag
*
* Description: This function is called to free the thread
* queue, and to kill all threads that were
* allocated as a result of the creation of the
* thread queue. If the immediate flag is set
* to non-zero, we will kill all threads
* immediately, otherwise we will wait for all
* threads to shutdown.
*
* Returns:
*/
void
{
if (!immediate) {
&to);
}
} /* end if */
/*
* shutdown condition, broadcast to all the threads waiting to
* terminate them
*/
} /* end while */
} /* end if */
}