library/src/blockclass/BP-BlockClass.cc
// This processor routine will create thread_n (count) threads, with startRoutine as an entry point.  The threads
// will take arguments directly from the thread_params pointer array.  New threads will be created as threads exit
// untill each element of thread_params has been completed.  After this routine has exited, you can check the individual
// parameter structures for information regarding their arbitrary input/output/statuses.
This processor routine will create thread_n (count) threads, with startRoutine as an entry point.  The threads
will take arguments directly from the thread_params pointer array.  New threads will be created as threads exit
untill each element of thread_params has been completed.  After this routine has exited, you can check the individual
parameter structures for information regarding their arbitrary input/output/statuses.
This processor routine will create thread_n ( count ) threads with startRoutine as an entry point . The threads will take arguments directly from the thread_params pointer array . New threads will be created as threads exit untill each element of thread_params has been completed . After this routine has exited you can check the individual parameter structures for information regarding their arbitrary input / output / statuses .
BP_ERROR_T Block::threadedDataProcessor(size_t thread_n, void *(startRoutine)(void*), void ** thread_params, size_t param_n)
{
	if(!thread_params)
	{
		this->AddErrorToStack("Error: Tried to run threaded data processor with no parameters set.");
		return ERR_FAILURE;
	}
	if(!startRoutine)
	{
		this->AddErrorToStack("Error: Tried to run the threaded data processor with no start routine set.");
		return ERR_FAILURE;
	}
	if(!thread_n)
	{
		this->AddErrorToStack("Error: Tried to run the threaded data processor without an arbitrary thread count (count is zero)");
		return ERR_FAILURE;
	}
	if(this->threads)
	{
		this->AddErrorToStack("Error: This block object seems to already have a this->threads structure.  Try destroying the existing thread configuration before dispatching this routine");
		return ERR_FAILURE;
	}
	if(!param_n)
	{
		printf("\n [+] Error using threaded data dispatcher.  You have a zero parameter count.");
		return ERR_FAILURE;
	}
	if(thread_n > param_n)
		thread_n = param_n;
	size_t n = 0;
	char name[128];
	for(n = 0; n < thread_n; n++)
	{
		memset(&name, 0x00, sizeof(name));
		sprintf((char *)&name, "auto-thread %u", n);
		this->CreateNewThread(startRoutine, thread_params[n], sizeof(void *), name);
	}
	size_t finished_processing = 0;
	P_THREAD_LIST thread_inner_iter = NULL;
	size_t data_index = 0;
	data_index = n;
	BP_ERROR_T thread_running_status = ERR_FAILURE;
	do
	{
		size_t threads_are_still_running = 0;
		size_t thread_cnt = 0;
		if(!this->threads)
		{
			printf("\n [ERR] No threads?");
			break;
		}
		for(
				thread_inner_iter = this->threads->first;
				thread_inner_iter;
				thread_inner_iter = thread_inner_iter->next
			)
		{
			while(!thread_params[data_index])
			{
				if(data_index >= param_n)
				{
					finished_processing = 1;
					break;
				}
				data_index++;
			}
			if(finished_processing)
				break;
			if(thread_inner_iter->threadState == BP_THREAD_STATUS_EXITED)
			{
				printf("\n Redispatching thread!!:%u, %p, %u",thread_inner_iter->threadId, thread_inner_iter->sharedData, data_index);
				printf("\n Current running threads?: %u", thread_inner_iter->thread_mgmt_data->total_running_threads);
				printf("\n");
				thread_inner_iter->sharedData = thread_params[data_index];
				this->redispatchThread(thread_inner_iter);
				data_index++;
			}
			if(this->thread_info.total_running_threads)
			{
				threads_are_still_running = 1;
			}
			if(data_index == param_n)
			{
				finished_processing = 1;
				break;
			}
		}
		if(threads_are_still_running)
		{
			usleep(40);
			int wait_status = 0;
			waitpid((pid_t) -1, &wait_status, 0);
			continue;
		}
		if(data_index == (param_n))
			finished_processing = 1;
	} while(!finished_processing);
	int status = 0;
	waitpid((pid_t) -1, &status, WNOHANG);
	while(this->thread_info.total_running_threads)
	{
		usleep(10);
	}
	if(this->thread_info.total_running_threads)
	{
		printf("\n Exiting with running threads?! %u", this->thread_info.total_running_threads);
	}
	RejoinAllThreads(this->threads);
	return ERR_SUCCESS;
}
BP_ERROR_T Block :: threadedDataProcessor ( size_t thread_n , void * ( startRoutine ) ( void * ) , void * * thread_params , size_t param_n ) { if ( ! thread_params ) { this -> AddErrorToStack ( " " ) ; return ERR_FAILURE ; } if ( ! startRoutine ) { this -> AddErrorToStack ( " " ) ; return ERR_FAILURE ; } if ( ! thread_n ) { this -> AddErrorToStack ( " " ) ; return ERR_FAILURE ; } if ( this -> threads ) { this -> AddErrorToStack ( " " ) ; return ERR_FAILURE ; } if ( ! param_n ) { printf ( " \n " ) ; return ERR_FAILURE ; } if ( thread_n > param_n ) thread_n = param_n ; size_t n = 0 ; char name [ 128 ] ; for ( n = 0 ; n < thread_n ; n ++ ) { memset ( & name , 0x00 , sizeof ( name ) ) ; sprintf ( ( char * ) & name , " " , n ) ; this -> CreateNewThread ( startRoutine , thread_params [ n ] , sizeof ( void * ) , name ) ; } size_t finished_processing = 0 ; P_THREAD_LIST thread_inner_iter = NULL ; size_t data_index = 0 ; data_index = n ; BP_ERROR_T thread_running_status = ERR_FAILURE ; do { size_t threads_are_still_running = 0 ; size_t thread_cnt = 0 ; if ( ! this -> threads ) { printf ( " \n " ) ; break ; } for ( thread_inner_iter = this -> threads -> first ; thread_inner_iter ; thread_inner_iter = thread_inner_iter -> next ) { while ( ! thread_params [ data_index ] ) { if ( data_index >= param_n ) { finished_processing = 1 ; break ; } data_index ++ ; } if ( finished_processing ) break ; if ( thread_inner_iter -> threadState == BP_THREAD_STATUS_EXITED ) { printf ( " \n " , thread_inner_iter -> threadId , thread_inner_iter -> sharedData , data_index ) ; printf ( " \n " , thread_inner_iter -> thread_mgmt_data -> total_running_threads ) ; printf ( " \n " ) ; thread_inner_iter -> sharedData = thread_params [ data_index ] ; this -> redispatchThread ( thread_inner_iter ) ; data_index ++ ; } if ( this -> thread_info . total_running_threads ) { threads_are_still_running = 1 ; } if ( data_index == param_n ) { finished_processing = 1 ; break ; } } if ( threads_are_still_running ) { usleep ( 40 ) ; int wait_status = 0 ; waitpid ( ( pid_t ) - 1 , & wait_status , 0 ) ; continue ; } if ( data_index == ( param_n ) ) finished_processing = 1 ; } while ( ! finished_processing ) ; int status = 0 ; waitpid ( ( pid_t ) - 1 , & status , WNOHANG ) ; while ( this -> thread_info . total_running_threads ) { usleep ( 10 ) ; } if ( this -> thread_info . total_running_threads ) { printf ( " \n " , this -> thread_info . total_running_threads ) ; } RejoinAllThreads ( this -> threads ) ; return ERR_SUCCESS ; }
This processor routine will create thread_n (count) threads, with startRoutine as an entry point.
This processor routine will create thread_n ( count ) threads with startRoutine as an entry point .
// do some quick parameter validation // Check to see if we have an already existing threads configuration.  If we do, exit immediately.  In order // to use this routine an empty thread configuration state must exist. // check to make sure we have parameters // should never try to start more threads than the parameters we already have // start initial threads to process front of data // holds name buffer (zeroed on loop) // create thread for each specified // zero out name // add auto thread label // create new thread using managed threads // flag used to exit the thread processing loop below // used to iterate thread items without destroying this->threads pointer. // data index used for redispatching individual threads as they exit // set index as the top current running thread // set the initial running status (indicates whether or not we still have threads running) // flag to indicate if flags are still running // current running thread count // walk the thread list starting from first position // do null checks // exit if the processing is finished // attempt to join the thread if possible (unnecessary, already done in redispatcher) // pthread_join(thread_inner_iter->threadId, NULL); // if a thread has exited, attempt to (logically) redispatch the thread with // a new data parameter from our parameter stack. // display redispatch message // reset shared data pointer from parameter index // redispatch the thread // increment the data index so that the next dispatch uses a new parameter set // check if threads are still runnin // set flag // Exit the loop if the index has matched our parameter count.  No new thread should be created as the // data has all been processed. // can't ever exit if threads are still running // check to see if we have any process exits in the case that // we're using this dispatcher with a sub process. // waitpid((pid_t) -1, &wait_status, WNOHANG); // exit in the case of reaching the final data index // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% // %%% Potential Zombie Cleanups %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% // set status to zero // call waitpid to clean up any zombies that might exist // wait for all threads to exit // printf("\n exiting! %u", this->thread_info.total_running_threads); // rejoin all the exited threads (frees memory) // return indicating success