gigablast/open-source-search-engine
1
0
Fork 0
mirror of https://github.com/gigablast/open-source-search-engine.git synced 2024-10-03 19:57:18 +03:00
Code Issues Projects Releases Wiki Activity
master
open-source-search-engine/Threads.cpp
Dmitry Smirnov b1ace63607 codespell: spelling corrections
2021-05-06 01:52:55 +10:00

3368 lines
106 KiB
C++
Raw Permalink Blame History

#include "gb-include.h"
#include "BigFile.h"
#include "Threads.h"
#include "Errno.h"
#include "Loop.h"
#include <sys/time.h>
#include <sys/resource.h>
#include <sys/types.h> // getuid()/pid_t/getpid()
#include <sys/wait.h> // waitpid()
#include "Rdb.h" // g_mergeUrgent
#include <sched.h> // clone()
//#include "Msg16.h" // g_pid g_ticker
#include "XmlDoc.h" // g_pid g_ticker
#include "Profiler.h"
#include "Stats.h"
#include "Process.h"
// try using pthreads again
//#define PTHREADS
// use these stubs so libplotter.a works
#ifndef PTHREADS
int pthread_mutex_lock (pthread_mutex_t *t ) { return 0; }
int pthread_mutex_unlock (pthread_mutex_t *t ) { return 0; }
#else
#include <pthread.h>
//pthread_attr_t s_attr;
#endif
// main process id (or thread id if using pthreads)
//static pid_t s_pid = (pid_t) -1;
// on 64-bit architectures pthread_t is 64 bit and pid_t is 32 bit:
static pthread_t s_pid = (pthread_t) -1;
//pid_t getpidtid() {
// on 64-bit architectures pthread_t is 64 bit and pid_t is 32 bit:
pthread_t getpidtid() {
#ifdef PTHREADS
// gettid() is a bit newer so not in our libc32...
// so kinda fake it. return the "thread" id, not process id.
// Threads::amThread() should still work based on thread ids because
// the main process has a unique thread id as well.
return pthread_self();
#else
return (pthread_t)getpid();
#endif
}
// BIG PROBLEM:
// When doing pthread_join it doesn't always ensure a thread doesn't go
// zombie. It seems like when SIGIOs are generated by sigqueue() because
// of a full signal queue, threads start going zombie on me.
// PROBLEM #1: with using the "state" to identify a thread in the queue.
// Often a caller make s a disk access using a certain "state" var.
// He gets control back when cleanUp() calls t->m_callback. And he launches
// another read thread in there, which calls Threads::exit(state) before
// the original thread entry's m_isOccupied is set to false. So
// Threads::exit(state) mistakenly picks the thread entry from the former
// thread picks it has the same state as its thread!
// PROBLEM #2: technically, a thread in Threads::exit() can set its m_done
// bit then send the signal. When the Loop sig handler sees the done bit is
// set it decrements the global thread count even though the thread may
// not have actually exited yet!! I spawned like 1,000 threads this way!!!!!
// a global class extern'd in .h file
Threads g_threads;
// . call this before calling a ThreadEntry's m_startRoutine
// . allows us to set the priority of the thread
int startUp ( void *state ) ;
void *startUp2 ( void *state ) ;
// JAB: warning abatement
//static void launchThreadsWrapper ( int fd , void *state );
static void killStalledFiltersWrapper ( int fd , void *state );
static void makeCallback ( ThreadEntry *t ) ;
// is the caller a thread?
bool Threads::amThread () {
if ( s_pid == (pthread_t)-1 ) return false;
#ifdef PTHREADS
// gettid() is a bit newer so not in our libc32...
// so kinda fake it. return the "thread" id, not process id.
// Threads::amThread() should still work based on thread ids because
// the main process has a unique thread id as well.
return (pthread_self() != s_pid);
#else
return ( getpidtid() != s_pid );
#endif
}
#ifndef PTHREADS
static int32_t s_bad = 0;
static int32_t s_badPid = -1;
#endif
#define MAX_PID 32767
#ifndef PTHREADS
static int s_errno ;
static int s_errnos [ MAX_PID + 1 ];
// this was improvised from linuxthreads/errno.c
//#define CURRENT_STACK_FRAME ({ char __csf; &__csf; })
// WARNING: you MUST compile with -DREENTRANT for this to work
int *__errno_location (void) {
int32_t pid = (int32_t) getpid();
//if ( pid == s_pid ) return &g_errno;
if ( pid <= (int32_t)MAX_PID ) return &s_errnos[pid];
s_bad++;
s_badPid = pid;
return &s_errno;
}
#endif
// stack must be page aligned for mprotect
#define THRPAGESIZE 8192
// how much of stack to use as guard space
#define GUARDSIZE (32*1024)
// . crashed in saving with 800k, so try 1M
// . must be multiple of THRPAGESIZE
//#define STACK_SIZE ((512+256) * 1024)
// pthread_create() cores in calloc() if we don't make STACK_SIZE bigger:
#define STACK_SIZE ((512+256+1024) * 1024)
// jeta was having some problems, but i don't think they were related to
// this stack size of 512k, but i will try boosting to 800k anyway.
//#define STACK_SIZE (512 * 1024)
// 256k was not enough, try 512k
//#define STACK_SIZE (256 * 1024)
// at 128k (and even 200k) some threads do not return! why?? there's
// obviously some stack overwriting going on!
//#define STACK_SIZE (128 * 1024)
static char *s_stackAlloc = NULL;
static int32_t s_stackAllocSize;
//#ifndef PTHREADS
static char *s_stack = NULL;
static int32_t s_stackSize;
static char *s_stackPtrs [ MAX_STACKS ];
//#endif
static int32_t s_next [ MAX_STACKS ];
static int32_t s_head ;
// returns NULL if none left
int32_t Threads::getStack ( ) {
if ( s_head == -1 ) return -1;
int32_t i = s_head;
s_head = s_next [ s_head ];
return i;
}
void Threads::returnStack ( int32_t si ) {
if ( s_head == -1 ) { s_head = si; s_next[si] = -1; return; }
s_next[si] = s_head;
s_head = si;
}
void Threads::reset ( ) {
if ( s_stackAlloc ) {
mprotect(s_stackAlloc, s_stackAllocSize, PROT_READ|PROT_WRITE);
mfree ( s_stackAlloc , s_stackAllocSize,
"ThreadStack");
}
s_stackAlloc = NULL;
for ( int32_t i = 0 ; i < MAX_THREAD_QUEUES ; i++ )
m_threadQueues[i].reset();
}
Threads::Threads ( ) {
m_numQueues = 0;
m_initialized = false;
}
void Threads::setPid ( ) {
// set s_pid to the main process id
#ifdef PTHREADS
// on 64bit arch pid is 32 bit and pthread_t is 64 bit
s_pid = (pthread_t)pthread_self();
//log(LOG_INFO,
// "threads: main process THREAD id = %"UINT32"",(int32_t unsigned)s_pid);
pthread_t tid = pthread_self();
sched_param param;
int policy;
// scheduling parameters of target thread
pthread_getschedparam ( tid, &policy, &param);
//log(LOG_INFO,
// "threads: min/max thread priority settings = %"INT32"/%"INT32" (policy=%"INT32")",
// (int32_t)sched_get_priority_min(policy),
// (int32_t)sched_get_priority_max(policy),
// (int32_t)policy);
#else
s_pid = getpid();
#endif
}
bool Threads::init ( ) {
if ( m_initialized ) return true;
m_initialized = true;
m_needsCleanup = false;
//m_needBottom = false;
// sanity check
//if ( sizeof(pthread_t) > sizeof(pid_t) ) { char *xx=NULL;*xx=0; }
if ( sizeof(pid_t) > sizeof(pthread_t) ) { char *xx=NULL;*xx=0; }
setPid();
#ifdef _STACK_GROWS_UP
return log("thread: Stack growing up not supported.");
#endif
//g_conf.m_logDebugThread = true;
// . damn, this only applies to fork() calls, i guess
// . a quick a dirty way to restrict # of threads so we don't explode
/*
struct rlimit lim;
lim.rlim_max = 100;
if ( setrlimit(RLIMIT_NPROC,&lim) )
log("thread::init: setrlimit: %s", mstrerror(errno) );
else
log("thread::init: set max number of processes to 100");
*/
// allow threads until disabled
m_disabled = false;
/*
// # of low priority threads launched and returned
m_hiLaunched = 0;
m_hiReturned = 0;
m_loLaunched = 0;
m_loReturned = 0;
*/
//int32_t m_queryMaxBigDiskThreads ; // > 1M read
//int32_t m_queryMaxMedDiskThreads ; // 100k - 1M read
//int32_t m_queryMaxSmaDiskThreads ; // < 100k per read
// categorize the disk read sizes by these here
// g_conf.m_bigReadSize = 0x7fffffff;
// g_conf.m_medReadSize = 1000000;
// g_conf.m_smaReadSize = 100000;
// . register a sleep wrapper to launch threads every 30ms
// . sometimes a bunch of threads mark themselves as done and the
// cleanUp() handler sees them as all still launched so it doesn't
// launch any new ones
//if ( ! g_loop.registerSleepCallback(30,NULL,launchThreadsWrapper))
// return log("thread: Failed to initialize timer callback.");
if ( ! g_loop.registerSleepCallback(1000,NULL,
killStalledFiltersWrapper,0))
return log("thread: Failed to initialize timer callback2.");
// debug
//log("thread: main process has pid=%"INT32"",(int32_t)s_pid);
// . set priority of the main process to 0
// . setpriority() only applies to SCHED_OTHER threads
// . priority of threads with niceness 0 will be 0
// . priority of threads with niceness 1 will be 10
// . priority of threads with niceness 2 will be 20
// . see 'man sched_setscheduler' for detail scheduling info
// . no need to call getpid(), 0 for pid means the current process
#ifndef PTHREADS
if ( setpriority ( PRIO_PROCESS, getpid() , 0 ) < 0 )
log("thread: Call to setpriority failed: %s.",
mstrerror(errno));
#endif
// make multiplier higher for raids, can do more seeks
//int32_t m = 1;
//#ifdef _LARS_
//m = 3;
//#endif
// register the types of threads here instead of in main.cpp
//if ( ! g_threads.registerType ( DISK_THREAD ,m*20/*maxThreads*/))
// try running blaster with 5 threads and you'll
// . see like a double degrade in performance for some reason!!
// . TODO: why?
// . well, this should be controlled g_conf.m_maxSpiderDiskThreads
// for niceness 1+ threads, and g_conf.m_maxPriorityDiskThreads for
// niceness 0 and below disk threads
// . 100 maxThreads out at a time, 32000 can be queued
if ( ! g_threads.registerType ( DISK_THREAD ,100/*maxThreads*/,32000))
return log("thread: Failed to register thread type." );
// . these are used by Msg5 to merge what it reads from disk
// . i raised it from 1 to 2 and got better response time from Msg10
// . i leave one open in case one is used for doing a big merge
// with high niceness cuz it would hold up high priority ones!
// . TODO: is there a better way? cancel it when UdpServer calls
// Threads::suspendLowPriorityThreads() ?
// . this used to be 2 but now defaults to 10 in Parms.cpp. i found
// i have less int32_t gray lines in the performance graph when i
// did that on trinity.
int32_t max2 = g_conf.m_maxCpuMergeThreads;
if ( max2 < 1 ) max2 = 1;
if ( ! g_threads.registerType ( MERGE_THREAD , max2,1000) )
return log("thread: Failed to register thread type." );
// will raising this from 1 to 2 make it faster too?
// i raised since global specs new servers have 2 (hyperthreaded?) cpus
int32_t max = g_conf.m_maxCpuThreads;
if ( max < 1 ) max = 1;
if ( ! g_threads.registerType ( INTERSECT_THREAD,max,10) )
return log("thread: Failed to register thread type." );
// filter thread spawned to call popen() to filter an http reply
if ( ! g_threads.registerType ( FILTER_THREAD, 2/*maxThreads*/,300) )
return log("thread: Failed to register thread type." );
// RdbTree uses this to save itself
if ( ! g_threads.registerType ( SAVETREE_THREAD,1/*maxThreads*/,100) )
return log("thread: Failed to register thread type." );
// . File.cpp spawns a rename thread for doing renames and unlinks
// . doing a tight merge on titldb can be ~250 unlinks
// . MDW up from 1 to 30 max, after doing a ddump on 3000+ collections
// it was taking forever to go one at a time through the unlink
// thread queue. seemed like a 1 second space between unlinks.
// 1/23/1014
if ( ! g_threads.registerType ( UNLINK_THREAD,5/*maxThreads*/,3000) )
return log("thread: Failed to register thread type." );
// generic multipurpose
if ( ! g_threads.registerType (GENERIC_THREAD,20/*maxThreads*/,100) )
return log("thread: Failed to register thread type." );
// for call SSL_accept() which blocks for 10ms even when socket
// is non-blocking...
//if (!g_threads.registerType (SSLACCEPT_THREAD,20/*maxThreads*/,100))
// return log("thread: Failed to register thread type." );
//#ifndef PTHREADS
// sanity check
if ( GUARDSIZE >= STACK_SIZE )
return log("thread: Stack guard size too big.");
// not more than this outstanding
int32_t maxThreads = 0;
for ( int32_t i = 0 ; i < m_numQueues ; i++ )
maxThreads += m_threadQueues[i].m_maxLaunched;
// limit to stack we got
if ( maxThreads > MAX_STACKS ) maxThreads = MAX_STACKS;
// allocate the stack space
s_stackAllocSize = STACK_SIZE * maxThreads + THRPAGESIZE ;
// clear stack to help check for overwrites
s_stackAlloc = (char *) mcalloc ( s_stackAllocSize , "ThreadStack" );
if ( ! s_stackAlloc )
return log("thread: Unable to allocate %"INT32" bytes for thread "
"stacks.", s_stackAllocSize);
log(LOG_INIT,"thread: Using %"INT32" bytes for %"INT32" thread stacks.",
s_stackAllocSize,maxThreads);
// align
s_stack = (char *)(((uint64_t)s_stackAlloc+THRPAGESIZE-1)&~(THRPAGESIZE-1));
// new size
s_stackSize = s_stackAllocSize - (s_stack - s_stackAlloc);
// protect the whole stack while not in use
if ( mprotect ( s_stack , s_stackSize , PROT_NONE ) )
log("thread: Call to mprotect failed: %s.",mstrerror(errno));
// test
//s_stack[0] = 1;
// init the linked list
for ( int32_t i = 0 ; i < MAX_STACKS ; i++ ) {
if ( i == MAX_STACKS - 1 ) s_next[i] = -1;
else s_next[i] = i + 1;
s_stackPtrs[i] = s_stack + STACK_SIZE * i;
}
s_head = 0;
// don't do real time stuff for now
return true;
//#else
// . keep stack size small, 128k
// . if we get problems, we'll increase this to 256k
// . seems like it grows dynamically from 4K to up to 2M as needed
//if ( pthread_attr_setstacksize ( &s_attr , (size_t)128*1024 ) )
// return log("thread: init: pthread_attr_setstacksize: %s",
// mstrerror(errno));
//pthread_attr_setschedparam ( &s_attr , PTHREAD_EXPLICIT_SCHED );
//pthread_attr_setscope ( &s_attr , PTHREAD_SCOPE_SYSTEM );
// return true;
//#endif
}
// all types should be registered in main.cpp before any threads launch
bool Threads::registerType ( char type , int32_t maxThreads , int32_t maxEntries ) {
// return false and set g_errno if no more room
if ( m_numQueues >= MAX_THREAD_QUEUES ) {
g_errno = EBUFTOOSMALL;
return log(LOG_LOGIC,"thread: registerType: Too many thread "
"queues");
}
// initialize the ThreadQueue class for this type
if ( ! m_threadQueues[m_numQueues].init(type,maxThreads,maxEntries))
return false;
// we got one more queue now
m_numQueues++;
return true;
}
int32_t Threads::getNumThreadsOutOrQueued() {
int32_t n = 0;
for ( int32_t i = 0 ; i < m_numQueues ; i++ ) {
// skip INTERSECT_THREAD, used to intersect termlists to
// resolve a query. i've seen these get stuck in an infinite
// loop sometimes.
if ( i == INTERSECT_THREAD ) continue;
// skip filter threads, those get stuck sometimes
if ( i == FILTER_THREAD ) continue;
// tally up all other threads
n += m_threadQueues[i].getNumThreadsOutOrQueued();
}
return n;
}
int32_t Threads::getNumWriteThreadsOut() {
return m_threadQueues[DISK_THREAD].getNumWriteThreadsOut();
}
int32_t Threads::getNumActiveWriteUnlinkRenameThreadsOut() {
// these do not countthreads that are done, and just awaiting join
int32_t n = m_threadQueues[DISK_THREAD].getNumWriteThreadsOut();
n += m_threadQueues[UNLINK_THREAD].getNumActiveThreadsOut();
return n;
}
// . returns false (and may set errno) if failed to launch a thread
// . returns true if thread added to queue successfully
// . may be launched instantly or later depending on # of threads in the queue
bool Threads::call ( char type ,
int32_t niceness ,
void *state ,
void (* callback )(void *state,ThreadEntry *t) ,
void *(* startRoutine)(void *state,ThreadEntry *t) ) {
// debug
//return false;
#ifdef _VALGRIND_
return false;
#endif
g_errno = 0;
// don't spawn any if disabled
if ( m_disabled ) return false;
if ( ! g_conf.m_useThreads ) return false;
if ( type == DISK_THREAD && ! g_conf.m_useThreadsForDisk )
return false;
if ( type == MERGE_THREAD && ! g_conf.m_useThreadsForIndexOps )
return false;
if ( type == INTERSECT_THREAD && ! g_conf.m_useThreadsForIndexOps )
return false;
if ( type == FILTER_THREAD && ! g_conf.m_useThreadsForSystemCalls )
return false;
if ( ! m_initialized && ! init() )
return log("db: Threads init failed." );
// . sanity check
// . a thread can NOT call this
//if ( getpid() != s_pid ) {
// fprintf(stderr,"thread: call: bad engineer\n");
// ::exit(-1);
//}
// don't launch for now
//return false;
// . sanity check
// . ensure top 4 bytes of state is the callback
//if ( *(int32_t *)state != (int32_t)callback ) {
// g_errno = EBADENGINEER;
// sleep(50000);
// return log("thread: call: top 4 bytes of state != callback");
//}
// debug msg
//log("adding thread to queue, type=%"INT32"",(int32_t)type);
// find the type
int32_t i;
for ( i = 0 ; i < m_numQueues ; i++ )
if ( m_threadQueues[i].m_threadType == type ) break;
// bitch if type not added via registerType() call
if ( i == m_numQueues ) {
g_errno = EBADENGINEER;
return log(LOG_LOGIC,"thread: addtoqueue: Unregistered "
"thread type %"INT32"",(int32_t)type);
}
// debug msg
//log("thread: call: adding entry for thread");
// . add to this queue
// . returns NULL and sets g_errno on error
ThreadEntry *t = m_threadQueues[i].addEntry(niceness,state,
callback,startRoutine);
if ( ! t ) return log("thread: Failed to add entry to thread pool: "
"%s.",mstrerror(g_errno));
// debug msg
//log("added");
// clear g_errno
//g_errno = 0;
// . try to launch as many threads as we can
// . this sets g_errno on error
// . if it has an error, just ignore it, our thread is queued
launchLoop:
if ( m_threadQueues[i].launchThread2 ( ) )
goto launchLoop;
//if ( ! m_threadQueues[i].launchThread2 ( t ) && g_errno ) {
// log("thread: failed thread launch: %s",mstrerror(g_errno));
// return false;
//}
// return false if there was an error launching the thread
//if ( g_errno ) return false;
// clear g_errno
g_errno = 0;
// success
return true;
}
// static void launchThreadsWrapper ( int fd , void *state ) {
// // debug msg
// //if ( g_conf.m_timingDebugEnabled )
// // log("thread: launchThreadsWrapper: entered");
// // clean up
// g_threads.cleanUp(NULL,1000);
// // and launch
// g_threads.launchThreads();
// }
static void killStalledFiltersWrapper ( int fd , void *state ) {
// bail if no pid
if ( g_pid == -1 ) return;
// . only kill after ticker reaches a count of 30
// . we are called once every second, so inc it each time
int32_t timeout = g_filterTimeout;
if ( timeout <= 0 ) timeout = 30;
if ( g_ticker++ < timeout ) return;
// debug
log("threads: killing stalled filter process of age %"INT32" "
"seconds and pid=%"INT32".",g_ticker,(int32_t)g_pid);
// kill him
int err = kill ( g_pid , 9 );
// don't kill again
g_pid = -1;
if ( err != 0 ) log("threads: kill filter: %s", mstrerror(err) );
}
// . called by g_loop in Loop.cpp after getting a SI_QUEUE signal that it
// is from when a thread exited
// . we put that signal there using sigqeueue() in Threads::exit()
// . this way another thread can be launched right away
int32_t Threads::launchThreads ( ) {
// stop launching threads if trying to exit.
// only launch save tree threads. so if in the middle of saving
// we allow it to complete?
if ( g_process.m_mode == EXIT_MODE )
return 0;
// try launching from each queue
int32_t numLaunched = 0;
// try to launch DISK threads last so cpu-based threads get precedence
for ( int32_t i = m_numQueues - 1 ; i >= 0 ; i-- ) {
// clear g_errno
g_errno = 0;
// launch as many threads as we can from queue #i
while ( m_threadQueues[i].launchThread2( ) )
numLaunched++;
// continue if no g_errno set
if ( ! g_errno ) continue;
// otherwise bitch about it
log("thread: Failed to launch thread: %s.",mstrerror(g_errno));
}
// clear g_errno
g_errno = 0;
return numLaunched;
}
// . will cancel currently running low priority threads
// . will prevent any low priority threads from launching
// . will only cancel disk threads for now
void Threads::suspendLowPriorityThreads() {
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: SUSPENDING LOW-PRIORITY THREADS.");
// just cancel disk threads for now
for ( int32_t i = 0 ; i < m_numQueues; i++ )
m_threadQueues[i].suspendLowPriorityThreads();
}
void Threads::resumeLowPriorityThreads() {
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: RESUMING LOW-PRIORITY THREADS.");
for ( int32_t i = 0 ; i < m_numQueues; i++ )
m_threadQueues[i].resumeLowPriorityThreads();
}
//void Threads::cancelLowPriorityThreads () {
// for ( int32_t i = 0 ; i < m_numQueues; i++ )
// m_threadQueues[i].cancelLowPriorityThreads();
//}
void checkList ( ThreadEntry **headPtr , ThreadEntry **tailPtr ) {
ThreadEntry *t = *headPtr;
// another check
if ( tailPtr && *headPtr && ! *tailPtr ) { char *xx=NULL;*xx=0; }
if ( tailPtr && ! *headPtr && *tailPtr ) { char *xx=NULL;*xx=0; }
if ( ! t ) return;
// head can not have a prev
if ( t->m_prevLink ) { char *xx=NULL;*xx=0; }
ThreadEntry *last = NULL;
for ( ; t ; t = t->m_nextLink ) {
if ( t->m_prevLink != last ) { char *xx=NULL;*xx=0; }
if ( last && last->m_nextLink != t ) { char *xx=NULL;*xx=0; }
last = t;
}
if ( ! tailPtr ) return;
t = *tailPtr;
// tail can not have a next
if ( t->m_nextLink ) { char *xx=NULL;*xx=0; }
last = NULL;
for ( ; t ; t = t->m_prevLink ) {
if ( t->m_nextLink != last ) { char *xx=NULL;*xx=0; }
if ( last && last->m_prevLink != t ) { char *xx=NULL;*xx=0; }
last = t;
}
}
// remove from a head-only linked list
void removeLink ( ThreadEntry **headPtr , ThreadEntry *t ) {
// if ( g_conf.m_logDebugThread )
// logf(LOG_DEBUG,"thread: removeLink [t=0x%"PTRFMT"]",
// (PTRTYPE)t);
if ( g_conf.m_logDebugThread )
checkList ( headPtr , NULL );
if ( t->m_prevLink ) t->m_prevLink->m_nextLink = t->m_nextLink;
else *headPtr = t->m_nextLink;
if ( t->m_nextLink ) t->m_nextLink->m_prevLink = t->m_prevLink;
t->m_nextLink = NULL;
t->m_prevLink = NULL;
if ( g_conf.m_logDebugThread )
checkList ( headPtr , NULL );
}
// MDW: verify this::::: and the one above!!!!!!!!!!!!!!!!
void removeLink2 ( ThreadEntry **headPtr ,
ThreadEntry **tailPtr ,
ThreadEntry *t ) {
// if ( g_conf.m_logDebugThread )
// logf(LOG_DEBUG,"thread: removeLink2 [t=0x%"PTRFMT"]",
// (PTRTYPE)t);
if ( g_conf.m_logDebugThread )
checkList ( headPtr , tailPtr );
if ( t->m_prevLink ) t->m_prevLink->m_nextLink = t->m_nextLink;
else *headPtr = t->m_nextLink;
if ( t->m_nextLink ) t->m_nextLink->m_prevLink = t->m_prevLink;
else *tailPtr = t->m_prevLink;
t->m_nextLink = NULL;
t->m_prevLink = NULL;
if ( g_conf.m_logDebugThread )
checkList ( headPtr , tailPtr );
}
// add to a head/tail linked list's tail
void addLinkToTail ( ThreadEntry **headPtr ,
ThreadEntry **tailPtr ,
ThreadEntry *t ) {
// if ( g_conf.m_logDebugThread )
// logf(LOG_DEBUG,"thread: addLinkToTail [t=0x%"PTRFMT"]",
// (PTRTYPE)t);
if ( g_conf.m_logDebugThread )
checkList ( headPtr , tailPtr );
if ( *tailPtr ) {
(*tailPtr)->m_nextLink = t;
t->m_nextLink = NULL;
t->m_prevLink = *tailPtr;
*tailPtr = t; // t is the new tail
}
else {
*headPtr = t;
*tailPtr = t;
t->m_nextLink = NULL;
t->m_prevLink = NULL;
}
if ( g_conf.m_logDebugThread )
checkList ( headPtr , tailPtr );
}
// add to a head/tail linked list's head
void addLinkToHead ( ThreadEntry **headPtr ,
ThreadEntry **tailPtr ,
ThreadEntry *t ) {
// if ( g_conf.m_logDebugThread )
// logf(LOG_DEBUG,"thread: addLinkToHead [t=0x%"PTRFMT"]",
// (PTRTYPE)t);
if ( g_conf.m_logDebugThread )
checkList ( headPtr , tailPtr );
if ( *headPtr ) {
(*headPtr)->m_prevLink = t;
t->m_prevLink = NULL;
t->m_nextLink = *headPtr;
*headPtr = t; // t is the new head
}
else {
*headPtr = t;
*tailPtr = t;
t->m_nextLink = NULL;
t->m_prevLink = NULL;
}
if ( g_conf.m_logDebugThread )
checkList ( headPtr , tailPtr );
}
// add to a head-only linked list
void addLink ( ThreadEntry **headPtr ,
ThreadEntry *t ) {
// if ( g_conf.m_logDebugThread )
// logf(LOG_DEBUG,"thread: addLink [t=0x%"PTRFMT"]",
// (PTRTYPE)t);
if ( g_conf.m_logDebugThread )
checkList ( headPtr , NULL );
if ( *headPtr ) {
(*headPtr)->m_prevLink = t;
t->m_prevLink = NULL;
t->m_nextLink = *headPtr;
*headPtr = t; // t is the new head
}
else {
*headPtr = t;
t->m_nextLink = NULL;
t->m_prevLink = NULL;
}
if ( g_conf.m_logDebugThread )
checkList ( headPtr , NULL );
}
///////////////////////////////////////////////////////////////////////
// functions for ThreadQueue
///////////////////////////////////////////////////////////////////////
ThreadQueue::ThreadQueue ( ) {
m_entries = NULL;
m_entriesSize = 0;
}
void ThreadQueue::reset ( ) {
if ( m_entries ) mfree ( m_entries , m_entriesSize , "Threads" );
m_entries = NULL;
m_top = 0;
}
bool ThreadQueue::init ( char threadType, int32_t maxThreads, int32_t maxEntries ) {
m_threadType = threadType;
m_launched = 0;
m_returned = 0;
m_maxLaunched = maxThreads;
// # of low priority threads launched and returned
/*
m_hiLaunched = 0;
m_hiReturned = 0;
m_mdLaunched = 0;
m_mdReturned = 0;
m_loLaunched = 0;
m_loReturned = 0;
// we now count write threads so we can limit that
m_writesLaunched = 0;
m_writesReturned = 0;
// these are for disk threads, which we now limit based on read sizes
m_hiLaunchedBig = 0;
m_hiReturnedBig = 0;
m_mdLaunchedBig = 0;
m_mdReturnedBig = 0;
m_loLaunchedBig = 0;
m_loReturnedBig = 0;
m_hiLaunchedMed = 0;
m_hiReturnedMed = 0;
m_mdLaunchedMed = 0;
m_mdReturnedMed = 0;
m_loLaunchedMed = 0;
m_loReturnedMed = 0;
m_hiLaunchedSma = 0;
m_hiReturnedSma = 0;
m_mdLaunchedSma = 0;
m_mdReturnedSma = 0;
m_loLaunchedSma = 0;
m_loReturnedSma = 0;
*/
//m_entriesUsed = 0;
//m_top = 0;
m_isLowPrioritySuspended = false;
// alloc space for entries
m_maxEntries = maxEntries;
m_entriesSize = sizeof(ThreadEntry)*m_maxEntries;
m_entries = (ThreadEntry *)mmalloc ( m_entriesSize , "Threads" );
if ( ! m_entries ) return log("thread: Failed to allocate %"INT32" bytes "
"for thread queue.",m_entriesSize);
// debug msg
//log("INIT CALLED. setting all m_isDone to 1.");
// clear m_isOccupied et al for new guys
//for ( int32_t i = 0 ; i < MAX_THREAD_ENTRIES ; i++ ) {
for ( int32_t i = 0 ; i < m_maxEntries ; i++ ) {
m_entries[i].m_isOccupied = false;
m_entries[i].m_isLaunched = false;
m_entries[i].m_isDone = true;
m_entries[i].m_qnum = threadType;
m_entries[i].m_stack = NULL;
}
m_emptyHead = NULL;
m_waitHead0 = NULL;
m_waitHead1 = NULL;
m_waitHead2 = NULL;
m_waitHead3 = NULL;
m_waitHead4 = NULL;
m_waitHead5 = NULL;
m_waitHead6 = NULL;
m_waitTail0 = NULL;
m_waitTail1 = NULL;
m_waitTail2 = NULL;
m_waitTail3 = NULL;
m_waitTail4 = NULL;
m_waitTail5 = NULL;
m_waitTail6 = NULL;
m_launchedHead = NULL;
// do not spam the log with log debug msgs even if it is on
char debug = g_conf.m_logDebugThread;
g_conf.m_logDebugThread = 0;
for ( int32_t i = 0 ; i < m_maxEntries ; i++ ) {
ThreadEntry *t = &m_entries[i];
t->m_prevLink = NULL;
t->m_nextLink = NULL;
addLink ( &m_emptyHead , t );
}
g_conf.m_logDebugThread = debug;
return true;
}
int32_t ThreadQueue::getNumActiveThreadsOut() {
int32_t n = 0;
for ( int32_t i = 0 ; i < m_maxEntries ; i++ ) {
ThreadEntry *e = &m_entries[i];
if ( ! e->m_isOccupied ) continue;
if ( ! e->m_isLaunched ) continue;
// if it is done and just waiting for a join, do not count
if ( e->m_isDone ) continue;
n++;
}
return n;
}
int32_t ThreadQueue::getNumThreadsOutOrQueued() {
// MDW: we also need to count threads that are returned but need their
// callback called so, in the case of RdbDump, the rdblist that was written
// to disk can update the rdbmap before it gets saved, so it doesn't get
// out of sync. Process.cpp calls .suspendMerge() to make sure that all
// merge operations are suspended as well.
int32_t n = 0;
for ( int32_t i = 0 ; i < m_maxEntries ; i++ ) {
ThreadEntry *e = &m_entries[i];
if ( e->m_isOccupied ) n++;
}
return n;
/*
int32_t n = m_launched - m_returned;
for ( int32_t i = 0 ; i < m_maxEntries ; i++ ) {
ThreadEntry *e = &m_entries[i];
if ( ! e->m_isOccupied ) continue;
if ( e->m_isLaunched ) continue;
if ( e->m_isDone ) continue;
// do not count "reads", only count writes
//if ( m_threadType == DISK_THREAD && e->m_state ) {
// FileState *fs = (FileState *)e->m_state;
// if ( fs->m_doWrite ) continue;
//}
}
return n;
*/
}
int32_t ThreadQueue::getNumWriteThreadsOut () {
// only consider disk threads
if ( m_threadType != DISK_THREAD ) return 0;
int32_t n = 0;
for ( int32_t i = 0 ; i < m_maxEntries ; i++ ) {
ThreadEntry *e = &m_entries[i];
if ( ! e->m_isOccupied ) continue;
if ( ! e->m_isLaunched ) continue;
if ( e->m_isDone ) continue;
FileState *fs = (FileState *)e->m_state;
if ( ! fs ) continue;
if ( ! fs->m_doWrite ) continue;
n++;
}
return n;
}
// return NULL and set g_errno on error
ThreadEntry *ThreadQueue::addEntry ( int32_t niceness ,
void *state ,
void (* callback )(void *state,
ThreadEntry *t) ,
void *(* startRoutine)(void *state,
ThreadEntry *t) ) {
// if we are 90% full and niceness is > 0, knock it off
// int32_t max = m_maxEntries;
// if ( m_threadType == DISK_THREAD && niceness > 0 ) {
// max = (m_maxEntries * 90) / 100;
// if ( max <= 0 ) max = 1;
// }
ThreadEntry *t = m_emptyHead;
if ( ! t ) {
g_errno = ENOTHREADSLOTS;
static time_t s_time = 0;
time_t now = getTime();
if ( now - s_time > 5 ) {
log("thread: Could not add thread to queue. Already "
"have %"INT32" entries.",m_maxEntries);
s_time = now;
}
return NULL;
}
// sanity
if ( t->m_isOccupied ) { char *xx=NULL;*xx=0; }
// debug test
//if ( rand() %10 == 1 ) { g_errno = ENOTHREADSLOTS; return NULL; }
// get first available entry, not in use
//int32_t i;
//for ( i = 0 ; i < MAX_THREAD_ENTRIES ; i++ )
// for ( i = 0 ; i < max ; i++ )
// if ( ! m_entries[i].m_isOccupied ) break;
// caution
// //if ( i >= MAX_THREAD_ENTRIES ) {
// if ( i >= max ) {
// g_errno = ENOTHREADSLOTS;
// static time_t s_time = 0;
// time_t now = getTime();
// if ( now - s_time > 5 ) {
// log("thread: Could not add thread to queue. Already "
// "have %"INT32" entries.",max);
// s_time = now;
// }
// return NULL;
// }
// debug msg
//fprintf(stderr,"addEntry my pid=%"UINT32"\n", (int32_t)getpid() );
// get an available entry
//ThreadEntry *t = &m_entries [ i ];
// debug msg
//log("claiming entry state=%"UINT32", occupied=%"INT32"",(int32_t)t->m_state,
// (int32_t)t->m_isOccupied);
// stick it in
t->m_niceness = niceness;
t->m_state = state;
t->m_callback = callback;
t->m_startRoutine = startRoutine;
t->m_isOccupied = true;
t->m_isCancelled = false;
t->m_stack = NULL;
// debug msg
//log("setting t=%"UINT32" m_isDone to 0", (int32_t)t );
t->m_isDone = false;
t->m_isLaunched = false;
t->m_queuedTime = gettimeofdayInMilliseconds();
t->m_readyForBail = false;
t->m_allocBuf = NULL;
t->m_allocSize = 0;
t->m_errno = 0;
t->m_bestHeadPtr = NULL;
t->m_bestTailPtr = NULL;
// and when the ohcrap callback gets called and the thread
// is cleaned up it will check the FileState readsize and
// m_isWrite to see which launch counts to decrement, so
// since FileState will be corrupted, we need to store
// this info directly into the thread entry.
if ( m_threadType == DISK_THREAD && t->m_state ) {
FileState *fs = (FileState *)t->m_state;
t->m_bytesToGo = fs->m_bytesToGo;
t->m_doWrite = fs->m_doWrite ;
}
else {
t->m_bytesToGo = 0;
t->m_doWrite = false;
}
// inc the used count
//m_entriesUsed++;
// debug msg
//log("m_entriesUsed now %"INT32"",m_entriesUsed);
// might have to inc top as well
//if ( i == m_top ) m_top++;
// there's only two linked lists we can wait in if we are not disk
if ( m_threadType != DISK_THREAD ) {
ThreadEntry **bestHeadPtr = NULL;
ThreadEntry **bestTailPtr = NULL;
if ( niceness <= 0 ) {
bestHeadPtr = &m_waitHead0;
bestTailPtr = &m_waitTail0;
}
// 'merge' threads from disk merge ops have niceness 1
else if ( niceness == 1 ) {
bestHeadPtr = &m_waitHead1;
bestTailPtr = &m_waitTail1;
}
// niceness is 2? MAX_NICENESS
else {
bestHeadPtr = &m_waitHead2;
bestTailPtr = &m_waitTail2;
}
// remove from empty list
removeLink ( &m_emptyHead , t );
// add to waiting list at the end
addLinkToTail ( bestHeadPtr , bestTailPtr , t );
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"queued %s thread for launch1. "
"niceness=%"INT32". ", (PTRTYPE)t,
getThreadType(), (int32_t)niceness );
return t;
}
//
// otherwise we are a disk thread, there's 7 linked lists to use
//
// shortcuts
//int32_t rs = t->m_bytesToGo;
char nice = t->m_niceness;
// get best thread candidate from best linked list of candidates
ThreadEntry **bestHeadPtr = NULL;
ThreadEntry **bestTailPtr = NULL;
// short/med/long high priority (niceness 0) disk reads in head0/1/2
// but we can't launch one more if already at our quota.
if ( ! bestHeadPtr && nice == 0 ) { //&& rs <= g_conf.m_smaReadSize ) {
bestHeadPtr = &m_waitHead0;
bestTailPtr = &m_waitTail0;
}
// if ( ! bestHeadPtr && nice == 0 && rs <= g_conf.m_medReadSize ) {
// bestHeadPtr = &m_waitHead1;
// bestTailPtr = &m_waitTail1;
// }
// if ( ! bestHeadPtr && nice == 0 ) {
// bestHeadPtr = &m_waitHead2;
// bestTailPtr = &m_waitTail2;
// }
// low priority (merge or dump) disk WRITES go in waithead4
if ( ! bestHeadPtr && t->m_doWrite ) {
bestHeadPtr = &m_waitHead4;
bestTailPtr = &m_waitTail4;
}
// niceness 1 read threads here
if ( ! bestHeadPtr && nice == 1 ) {
bestHeadPtr = &m_waitHead5;
bestTailPtr = &m_waitTail5;
}
// niceness 2 read threads here
if ( ! bestHeadPtr ) {
bestHeadPtr = &m_waitHead6;
bestTailPtr = &m_waitTail6;
}
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"remove from empty list, add to wait list",
(PTRTYPE)t);
// remove from empty list
removeLink ( &m_emptyHead , t );
// sanity
if ( m_emptyHead == t ) { char *xx=NULL;*xx=0; }
// add to the new waiting list at the end
addLinkToTail ( bestHeadPtr , bestTailPtr , t );
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"queued %s thread for launch2. "
"niceness=%"INT32". ", (PTRTYPE)t,
getThreadType(), (int32_t)niceness );
// success
return t;
}
int32_t Threads::timedCleanUp (int32_t maxTime, int32_t niceness) {
// skip it if exiting
if ( g_process.m_mode == EXIT_MODE )
return 0;
if ( ! m_needsCleanup ) return 0;
//if ( g_inSigHandler ) return 0;
int64_t startTime = gettimeofdayInMillisecondsLocal();
int64_t took = 0;
if ( niceness >= MAX_NICENESS ) m_needsCleanup = false;
//for ( int32_t i = -1 ; i <= niceness ; i++ ) {
for ( int32_t i = 0 ; i <= niceness ; i++ ) {
for ( int32_t j = 0 ; j < m_numQueues ; j++ )
m_threadQueues[j].timedCleanUp ( niceness );
launchThreads();
if ( maxTime < 0 ) continue;
took = startTime - gettimeofdayInMillisecondsLocal();
if ( took <= maxTime ) continue;
// ok, we have to cut if short...
m_needsCleanup = true;
break;
}
return took;
}
bool Threads::isHittingFile ( BigFile *bf ) {
return m_threadQueues[DISK_THREAD].isHittingFile(bf);
}
bool ThreadQueue::isHittingFile ( BigFile *bf ) {
// loop through candidates
for ( int32_t i = 0 ; i < m_top; i++ ) {
// point to it
ThreadEntry *t = &m_entries[i];
// must be occupied to be done (sanity check)
if ( ! t->m_isOccupied ) continue;
// must not be done
if ( t->m_isDone ) continue;
// must be launched.. really??
//if ( ! t->m_isLaunched ) continue;
// must be a read
if ( t->m_startRoutine != readwriteWrapper_r ) continue;
// int16_tcut
FileState *fs = (FileState *)t->m_state;
// get bigfile ptr
if ( fs->m_this == bf ) return true;
}
return false;
}
void Threads::bailOnReads ( ) {
m_threadQueues[DISK_THREAD].bailOnReads();
}
// Process.cpp calls these callbacks before their time in order to
// set EDISKSTUCK
void ThreadQueue::bailOnReads ( ) {
// note it
log("threads: bypassing read threads");
ThreadEntry *t = m_launchedHead;
ThreadEntry *nextLink = NULL;
// loop through candidates
//for ( int32_t i = 0 ; i < m_top; i++ ) {
for ( ; t ; t = nextLink ) {
// do it here in case we modify the linked list below
nextLink = t->m_nextLink;
// must be occupied to be done (sanity check)
if ( ! t->m_isOccupied ) continue;
// skip if not launched yet
//if ( ! t->m_isLaunched ) continue;
// must be niceness 0
if ( t->m_niceness != 0 ) continue;
// must not be done
if ( t->m_isDone ) continue;
// must not have already called callback
if ( t->m_callback == ohcrap ) continue;
// must be a read
if ( t->m_startRoutine != readwriteWrapper_r ) continue;
// int16_tcut
FileState *fs = (FileState *)t->m_state;
// do not stop writes
if ( fs->m_doWrite ) continue;
// must be niceness 0 too!
if ( fs->m_niceness != 0 ) continue;
// what is this? unlaunched...
//if ( t->m_pid == 0 ) continue;
// can only bail on a thread after it copies its FileState
// class into its stack so we can bypass it and free the
// original FileState without causing a core. if thread
// is not yet launched we have to call the callback here too
// otherwise it never gets launched until the disk is unstuck!
if ( ! t->m_readyForBail && t->m_isLaunched ) continue;
// set error
t->m_errno = EDISKSTUCK;
// set this too
g_errno = EDISKSTUCK;
// do not allow caller to free the alloc'd buf in case
// its read finally comes through!
t->m_allocBuf = fs->m_allocBuf;
t->m_allocSize = fs->m_allocSize;
fs->m_allocBuf = NULL;
fs->m_allocSize = 0;
// call it
t->m_callback ( t->m_state , t );
// do not re-call it...
t->m_callback = ohcrap;
// invalidate state (FileState usually)
t->m_state = NULL;
// do not launch if not yet launched
if ( t->m_isLaunched ) continue;
// delete him if not yet launched, otherwise we try to
// launch it later with a corrupted/unstable FileState...
// and that causes our launch counts to get out of whack i
// think...
t->m_isOccupied = false;
// note it
log("threads: bailing unlaunched thread");
// remove from linked list then
removeLink ( &m_launchedHead , t );
addLink ( &m_emptyHead , t );
// do we have to decrement top
// if ( m_top == i + 1 )
// while (m_top > 0 && ! m_entries[m_top-1].m_isOccupied)
// m_top--;
}
}
// BigFile.cpp's readwriteWrapper_r() ThreadEntry::m_callback gets set to
// ohcrap() because it was taking too long to do its read and we prematurely
// called its callback above in bailOnReads(). In that case we still have to
// free the disk read buffer which was never used. And doneWrapper() in
// BigFile.cpp is never called.
void ohcrap ( void *state , ThreadEntry *t ) {
// free the read buffer here then
if ( t->m_allocBuf )
mfree ( t->m_allocBuf , t->m_allocSize , "RdbScan" );
log("threads: got one");
}
// . cleans up any threads that have exited
// . their m_isDone should be set to true
// . don't process threads whose niceness is > maxNiceness
// . return true if we cleaned one up
bool ThreadQueue::timedCleanUp ( int32_t maxNiceness ) {
// top:
int32_t numCallbacks = 0;
ThreadEntry *t = m_launchedHead;
ThreadEntry *nextLink = NULL;
// loop through candidates
for ( ; t ; t = nextLink ) {
// get it here in case we remove 't' from the linked list below
nextLink = t->m_nextLink;
// point to it
//ThreadEntry *t = &m_entries[i];
// skip if not qualified
if ( t->m_niceness > maxNiceness ) continue;
// must be occupied to be done (sanity check)
if ( ! t->m_isOccupied ) continue;
// skip if not launched yet
if ( ! t->m_isLaunched ) continue;
// . we were segfaulting right here before because the thread
// was setting t->m_pid and at this point it was not
// set so t->m_pid was a bogus value
// . thread may have seg faulted, in which case sigbadhandler()
// in Loop.cpp will get it and set errno to 0x7fffffff
#ifndef PTHREADS
// MDW: i hafta take this out because the errno_location thing
// is not working on the newer gcc
if ( ! t->m_isDone && t->m_pid >= 0 &&
s_errnos [t->m_pid] == 0x7fffffff ) {
log("thread: Got abnormal thread termination. Seems "
"like the thread might have cored.");
s_errnos[t->m_pid] = 0;
goto again;
}
#endif
// skip if not done yet
if ( ! t->m_isDone ) continue;
#ifdef PTHREADS
// if pthread_create() failed it returns the errno and we
// needsJoin is false, so do not try to join
// to a thread if we did not create it, lest pthread_join()
// cores
if ( t->m_needsJoin ) {
// . join up with that thread
// . damn, sometimes he can block forever on his
// call to sigqueue(),
int64_t startTime = gettimeofdayInMillisecondsLocal();
int64_t took;
int32_t status = pthread_join ( t->m_joinTid , NULL );
took = startTime - gettimeofdayInMillisecondsLocal();
if ( took > 50 ) {
log("threads: pthread_join took %i ms",
(int)took);
}
if ( status != 0 ) {
log("threads: pthread_join %"INT64" = %s (%"INT32")",
(int64_t)t->m_joinTid,mstrerror(status),
status);
}
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: joined1 with "
"t=0x%"PTRFMT" "
"jointid=0x%"XINT64".",
(PTRTYPE)t,(int64_t)t->m_joinTid);
// re-protect this stack
mprotect ( t->m_stack + GUARDSIZE ,
STACK_SIZE - GUARDSIZE,
PROT_NONE );
g_threads.returnStack ( t->m_si );
t->m_stack = NULL;
}
#else
again:
int status ;
pid_t pid = waitpid ( t->m_pid , &status , 0 );
int err = errno;
// debug the waitpid
if ( g_conf.m_logDebugThread || g_process.m_exiting )
log(LOG_DEBUG,"thread: Waiting for t=0x%"PTRFMT" "
"pid=%"INT32".",
(PTRTYPE)t,(int32_t)t->m_pid);
// bitch and continue if join failed
if ( pid != t->m_pid ) {
// waitpid() gets interrupted by various signals so
// we need to repeat (SIGCHLD?)
if ( err == EINTR ) goto again;
log("thread: Call to waitpid(%"INT32") returned %"INT32": %s.",
(int32_t)t->m_pid,(int32_t)pid,mstrerror(err));
continue;
}
// if status not 0 then process got abnormal termination
if ( ! WIFEXITED(status) ) {
if ( WIFSIGNALED(status) )
log("thread: Child process (pid=%i) exited "
"from unhandled signal number %"INT32".",
pid,(int32_t)WTERMSIG(status));
else
log("thread: Child process (pid=%i) exited "
"for unknown reason." , pid );
}
//mfree ( t->m_stack , STACK_SIZE , "Threads" );
// re-protect this stack
mprotect ( t->m_stack + GUARDSIZE , STACK_SIZE - GUARDSIZE,
PROT_NONE );
g_threads.returnStack ( t->m_si );
t->m_stack = NULL;
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: joined with pid=%"INT32" pid=%"INT32".",
(int32_t)t->m_pid,(int32_t)t->m_pid);
#endif
// we may get cleaned up and re-used and our niceness reassignd
// right after set m_isDone to true, so save niceness
//int32_t niceness = t->m_niceness;
char qnum = t->m_qnum;
ThreadQueue *tq = &g_threads.m_threadQueues[(int)qnum];
if ( tq != this ) { char *xx = NULL; *xx = 0; }
// get read size before cleaning it up -- it could get nuked
//int32_t rs = 0;
bool isWrite = false;
if ( tq->m_threadType == DISK_THREAD ) { // && t->m_state ) {
//FileState *fs = (FileState *)t->m_state;
//rs = t->m_bytesToGo;
isWrite = t->m_doWrite ;
}
/*
if ( niceness <= 0) tq->m_hiReturned++;
else if ( niceness == 1) tq->m_mdReturned++;
else if ( niceness >= 2) tq->m_loReturned++;
// deal with the tiers for disk threads based on read sizes
if ( tq->m_threadType == DISK_THREAD ) {
// writes are special cases
if ( isWrite ) m_writesReturned++;
if ( rs >= 0 && niceness >= 2 ) {
if ( rs > g_conf.m_medReadSize )
tq->m_loReturnedBig++;
else if ( rs > g_conf.m_smaReadSize )
tq->m_loReturnedMed++;
else
tq->m_loReturnedSma++;
}
else if ( rs >= 0 && niceness >= 1 ) {
if ( rs > g_conf.m_medReadSize )
tq->m_mdReturnedBig++;
else if ( rs > g_conf.m_smaReadSize )
tq->m_mdReturnedMed++;
else
tq->m_mdReturnedSma++;
}
else if ( rs >= 0 ) {
if ( rs > g_conf.m_medReadSize )
tq->m_hiReturnedBig++;
else if ( rs > g_conf.m_smaReadSize )
tq->m_hiReturnedMed++;
else
tq->m_hiReturnedSma++;
}
}
*/
// now count him as returned
m_returned++;
// prepare for relaunch if we were cancelled
if ( t->m_isCancelled ) {
t->m_isCancelled = false;
t->m_isLaunched = false;
t->m_isDone = false;
// take out of launched linked list
removeLink ( &m_launchedHead , t );
// get the waiting linked list from whence we came
ThreadEntry **bestHeadPtr = t->m_bestHeadPtr;
ThreadEntry **bestTailPtr = t->m_bestTailPtr;
// put BACK into the waiting linked list at the head
addLinkToHead ( bestHeadPtr , bestTailPtr , t );
log("thread: Thread cancelled. Preparing thread "
"for relaunch");
continue;
}
numCallbacks++;
// not running any more
t->m_isLaunched = false;
// not occupied any more
t->m_isOccupied = false;
// do we have to decrement top
// if ( m_top == i + 1 )
// while (m_top > 0 && ! m_entries[m_top-1].m_isOccupied)
// m_top--;
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"remove from launched list, add to empty list",
(PTRTYPE)t);
// now move it to the empty list
removeLink ( &m_launchedHead , t );
addLink ( &m_emptyHead , t );
// send a cancel sig to the thread in case it's still there
//int err = pthread_cancel ( t->m_tid );
//if ( err != 0 ) log("thread: cleanUp: pthread_cancel: %s",
// mstrerror(err) );
// one less entry occupied
//m_entriesUsed--;
// debug msg
//log("m_entriesUsed now %"INT32"",m_entriesUsed);
// one more returned
//m_returned++;
// clear the g_errno in case set by a previous callback
//g_errno = 0;
// launch as many threads as we can before calling the
// callback since this may hog the CPU like Msg20 does
//g_threads.launchThreads();
g_errno = 0;
//g_loop.startBlockedCpuTimer();
//only allow a quickpoll if we are nice.
//g_loop.canQuickPoll(t->m_niceness);
makeCallback ( t );
//int64_t took = gettimeofdayInMilliseconds()-startTime;
//if(took > 8 && maxNiceness > 0) {
// if(g_conf.m_sequentialProfiling)
// log(LOG_TIMING,
// "admin: Threads spent %"INT64" ms to callback "
// "%"INT32" callbacks, nice: %"INT32"",
// took, numCallbacks, maxNiceness);
// g_threads.m_needBottom = true;
// maxNiceness = 0;
//}
// clear errno again
g_errno = 0;
if ( g_conf.m_logDebugThread ) {
int64_t now = gettimeofdayInMilliseconds();
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] %s done1. "
"active=%"INT32" "
"time since queued = %"UINT64" ms "
"time since launch = %"UINT64" ms "
"time since pre-exit = %"UINT64" ms "
"time since exit = %"UINT64" ms",
(PTRTYPE)t,
getThreadType() ,
(int32_t)(m_launched - m_returned) ,
(uint64_t)(now - t->m_queuedTime),
(uint64_t)(now - t->m_launchedTime),
(uint64_t)(now - t->m_preExitTime) ,
(uint64_t)(now - t->m_exitTime) );
}
}
//since we need finer grained control in loop, we no longer collect
//the callbacks, sort, then call them. we now call them right away
//that way we can break out if we start taking too long and
//give control back to udpserver.
return numCallbacks != 0;
}
void makeCallback ( ThreadEntry *t ) {
// sanity check - if in a high niceness callback, we should
// only be calling niceness 0 callbacks here
// no, this is only called from sleep wrappers originating from
// Loop.cpp, so we should be ok
//if ( g_niceness==0 && t->m_niceness ) { char *xx=NULL;*xx=0; }
// save it
int32_t saved = g_niceness;
// log it now
if ( g_conf.m_logDebugLoop || g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: enter thread callback t=0x%"PTRFMT" "
//"type=%s "
"state=0x%"PTRFMT" "
"nice=%"INT32"",
(PTRTYPE)t,
//getThreadType(),
(PTRTYPE)t->m_state,
(int32_t)t->m_niceness);
// time it?
int64_t start;
if ( g_conf.m_maxCallbackDelay >= 0 )
start = gettimeofdayInMillisecondsLocal();
// then set it
if ( t->m_niceness >= 1 ) g_niceness = 1;
else g_niceness = 0;
t->m_callback ( t->m_state , t );
// time it?
if ( g_conf.m_maxCallbackDelay >= 0 ) {
int64_t elapsed = gettimeofdayInMillisecondsLocal() - start;
if ( elapsed >= g_conf.m_maxCallbackDelay )
log("threads: Took %"INT64" ms to call "
"thread callback niceness=%"INT32"",
elapsed,(int32_t)saved);
}
// log it now
if ( g_conf.m_logDebugLoop || g_conf.m_logDebugThread )
log(LOG_DEBUG,"loop: exit thread callback t=0x%"PTRFMT" "
//"type=%s "
"nice=%"INT32"",
(PTRTYPE)t,
//getThreadType(),
(int32_t)t->m_niceness);
// restore global niceness
g_niceness = saved;
}
bool Threads::cleanUp ( ThreadEntry *t , int32_t maxNiceness ) {
bool didSomething = false;
loop:
// assume no more cleanup needed
m_needsCleanup = false;
//m_needBottom = false;
// debug msg
//log("cleanUp");
// debug msg
//log("cleaning up exited threads and calling callbacks");
for ( int32_t i = 0 ; i < m_numQueues ; i++ ) {
didSomething |= m_threadQueues[i].cleanUp( t , maxNiceness );
// . if we broke from the loop
//if(m_needBottom) maxNiceness = 0;
}
// . loop more if we got a new one
// . thread will set this when about to exit
// . waitpid() may be interrupted by a SIGCHLD and not get his pid
if ( m_needsCleanup ) goto loop;
return didSomething;
}
// . cleans up any threads that have exited
// . their m_isDone should be set to true
// . don't process threads whose niceness is > maxNiceness
bool ThreadQueue::cleanUp ( ThreadEntry *tt , int32_t maxNiceness ) {
// call all callbacks after all threads are cleaned up
void (* callbacks[64])(void *state,ThreadEntry *);
void *states [64];
int64_t times [64];
int64_t times2 [64];
int64_t times3 [64];
int64_t times4 [64];
ThreadEntry *tids [64];
int64_t startTime = gettimeofdayInMilliseconds();
// top:
ThreadEntry *t = m_launchedHead;
ThreadEntry *nextLink = NULL;
int32_t numCallbacks = 0;
// loop through candidates
//for ( int32_t i = 0 ; i < m_top && numCallbacks < 64 ; i++ ) {
for ( ; t && numCallbacks < 64 ; t = nextLink ) {
// do it here in case we modify the linked list below
nextLink = t->m_nextLink;
// point to it
//ThreadEntry *t = &m_entries[i];
// skip if not qualified
if ( t->m_niceness > maxNiceness ) {
//if(t->m_isDone) {
// g_threads.m_needBottom = true;
// //g_threads.m_needsCleanup = true;
//}
continue;
}
// must be occupied to be done (sanity check)
if ( ! t->m_isOccupied ) continue;
// skip if not launched yet
if ( ! t->m_isLaunched ) continue;
// . we were segfaulting right here before because the thread
// was setting t->m_pid and at this point it was not
// set so t->m_pid was a bogus value
// . thread may have seg faulted, in which case sigbadhandler()
// in Loop.cpp will get it and set errno to 0x7fffffff
#ifndef PTHREADS
// MDW: i hafta take this out because the errno_location thing
// is not working on the newer gcc
if ( ! t->m_isDone && t->m_pid >= 0 &&
s_errnos [t->m_pid] == 0x7fffffff ) {
log("thread: Got abnormal thread termination. Seems "
"like the thread might have cored.");
s_errnos[t->m_pid] = 0;
goto again;
}
#endif
// skip if not done yet
if ( ! t->m_isDone ) continue;
#ifdef PTHREADS
if ( t->m_needsJoin ) {
// . join up with that thread
// . damn, sometimes he can block forever on his
// call to sigqueue(),
int32_t status = pthread_join ( t->m_joinTid , NULL );
if ( status != 0 ) {
log("threads: "
"pthread_join2 %"INT64" = %s (%"INT32")",
(int64_t)t->m_joinTid,mstrerror(status),
status);
}
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: joined2 with "
"t=0x%"PTRFMT" "
"jointid=0x%"XINT64".",
(PTRTYPE)t,(int64_t)t->m_joinTid);
// re-protect this stack
mprotect ( t->m_stack + GUARDSIZE ,
STACK_SIZE - GUARDSIZE,
PROT_NONE );
g_threads.returnStack ( t->m_si );
t->m_stack = NULL;
}
#else
again:
int status ;
pid_t pid = waitpid ( t->m_pid , &status , 0 );
int err = errno;
// debug the waitpid
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: Waiting for "
"t=0x%"PTRFMT" pid=%"INT32".",
(PTRTYPE)t,(int32_t)t->m_pid);
// bitch and continue if join failed
if ( pid != t->m_pid ) {
// waitpid() gets interrupted by various signals so
// we need to repeat (SIGCHLD?)
if ( err == EINTR ) goto again;
log("thread: Call to waitpid(%"INT32") returned %"INT32": %s.",
(int32_t)t->m_pid,(int32_t)pid,mstrerror(err));
continue;
}
// if status not 0 then process got abnormal termination
if ( ! WIFEXITED(status) ) {
if ( WIFSIGNALED(status) )
log("thread: Child process (pid=%i) exited "
"from unhandled signal number %"INT32".",
pid,(int32_t)WTERMSIG(status));
else
log("thread: Child process (pid=%i) exited "
"for unknown reason." , pid );
}
//mfree ( t->m_stack , STACK_SIZE , "Threads" );
// re-protect this stack
mprotect ( t->m_stack + GUARDSIZE , STACK_SIZE - GUARDSIZE,
PROT_NONE );
g_threads.returnStack ( t->m_si );
t->m_stack = NULL;
#endif
// debug msg
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: joined with pid=%"INT32" pid=%"INT32".",
(int32_t)t->m_pid,(int32_t)t->m_pid);
// we may get cleaned up and re-used and our niceness reassignd
// right after set m_isDone to true, so save niceness
//int32_t niceness = t->m_niceness;
char qnum = t->m_qnum;
ThreadQueue *tq = &g_threads.m_threadQueues[(int)qnum];
if ( tq != this ) { char *xx = NULL; *xx = 0; }
/*
// get read size before cleaning it up -- it could get nuked
int32_t rs = 0;
bool isWrite = false;
if ( tq->m_threadType == DISK_THREAD ) { // && t->m_state ) {
//FileState *fs = (FileState *)t->m_state;
rs = t->m_bytesToGo;
isWrite = t->m_doWrite ;
}
if ( niceness <= 0) tq->m_hiReturned++;
else if ( niceness == 1) tq->m_mdReturned++;
else if ( niceness >= 2) tq->m_loReturned++;
// deal with the tiers for disk threads based on read sizes
if ( tq->m_threadType == DISK_THREAD ) {
// writes are special cases
if ( isWrite ) m_writesReturned++;
if ( rs >= 0 && niceness >= 2 ) {
if ( rs > g_conf.m_medReadSize )
tq->m_loReturnedBig++;
else if ( rs > g_conf.m_smaReadSize )
tq->m_loReturnedMed++;
else
tq->m_loReturnedSma++;
}
else if ( rs >= 0 && niceness >= 1 ) {
if ( rs > g_conf.m_medReadSize )
tq->m_mdReturnedBig++;
else if ( rs > g_conf.m_smaReadSize )
tq->m_mdReturnedMed++;
else
tq->m_mdReturnedSma++;
}
else if ( rs >= 0 ) {
if ( rs > g_conf.m_medReadSize )
tq->m_hiReturnedBig++;
else if ( rs > g_conf.m_smaReadSize )
tq->m_hiReturnedMed++;
else
tq->m_hiReturnedSma++;
}
}
*/
// . we should count down here, not in the master thread
// . solves Problem #2 ?
// . TODO: this is not necessaruly atomic we should set
// t->m_aboutToExit to true so cleanUp can periodically set
// m_returned to what it should be!!!
//g_threads.m_threadQueues[qnum].m_returned++;
// now count him as returned
m_returned++;
// prepare for relaunch if we were cancelled
if ( t->m_isCancelled ) {
t->m_isCancelled = false;
t->m_isLaunched = false;
t->m_isDone = false;
// take out of the linked list of launched threads
removeLink ( &m_launchedHead , t );
// what waiting linked list did it come from?
// get the waiting linked list from whence we came
ThreadEntry **bestHeadPtr = t->m_bestHeadPtr;
ThreadEntry **bestTailPtr = t->m_bestTailPtr;
// and put it back at the head
addLinkToHead ( bestHeadPtr , bestTailPtr , t );
log("thread: Thread cancelled. Preparing thread "
"for relaunch");
continue;
}
// debug msg
//log("[%"UINT32"] CLEANING UP THREAD type=%"INT32", numLaunched=%"INT32"",
// m_entries[i].m_tid , m_threadType , m_launched );
// remove it
// debug msg
//log("CLN TID=%"UINT32" t=%"UINT32"",(int32_t)t->m_tid , (int32_t)t);
//log("thread callback for tid=%"UINT32"",(int32_t)t->m_tid );
// . save important stuff before freeing up the ThreadEntry
// for possible take over.
// . calling the callback may launch a thread which may
// claim THIS thread entry, t
//void (* callback)(void *state);
//callback = t->m_callback;
//void *state = t->m_state;
callbacks [ numCallbacks ] = t->m_callback;
states [ numCallbacks ] = t->m_state;
times [ numCallbacks ] = t->m_queuedTime;
times2 [ numCallbacks ] = t->m_launchedTime;
times3 [ numCallbacks ] = t->m_preExitTime;
times4 [ numCallbacks ] = t->m_exitTime;
tids [ numCallbacks ] = t;
numCallbacks++;
// SOLUTION: before calling the callback which may launch
// another thread with this same tid, thus causing an error,
// we should set these to false first:
// not running any more
t->m_isLaunched = false;
// not occupied any more
t->m_isOccupied = false;
// do we have to decrement top
//if ( m_top == i + 1 )
// while (m_top > 0 && ! m_entries[m_top-1].m_isOccupied)
// m_top--;
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"remove from launched list, add to empty list",
(PTRTYPE)t);
// now move it to the empty list
removeLink ( &m_launchedHead , t );
addLink ( &m_emptyHead , t );
// send a cancel sig to the thread in case it's still there
//int err = pthread_cancel ( t->m_tid );
//if ( err != 0 ) log("thread: cleanUp: pthread_cancel: %s",
// mstrerror(err) );
// one less entry occupied
//m_entriesUsed--;
// debug msg
//log("m_entriesUsed now %"INT32"",m_entriesUsed);
// one more returned
//m_returned++;
// clear the g_errno in case set by a previous callback
//g_errno = 0;
// launch as many threads as we can before calling the
// callback since this may hog the CPU like Msg20 does
//g_threads.launchThreads();
g_errno = 0;
makeCallback ( t );
// int64_t took = gettimeofdayInMilliseconds()-startTime;
// if(took > 8 && maxNiceness > 0) {
// if(g_conf.m_sequentialProfiling)
// log(LOG_TIMING,
// "admin: Threads spent %"INT64" ms to callback "
// "%"INT32" callbacks, nice: %"INT32"",
// took, numCallbacks, maxNiceness);
// g_threads.m_needBottom = true;
// maxNiceness = 0;
// }
// clear errno again
g_errno = 0;
if ( g_conf.m_logDebugThread ) {
int64_t now = gettimeofdayInMilliseconds();
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] %s done2. "
"active=%"INT32" "
"time since queued = %"UINT64" ms "
"time since launch = %"UINT64" ms "
"time since pre-exit = %"UINT64" ms "
"time since exit = %"UINT64" ms",
(PTRTYPE)t,
getThreadType() ,
(int32_t)(m_launched - m_returned) ,
(uint64_t)(now - t->m_queuedTime),
(uint64_t)(now - t->m_launchedTime),
(uint64_t)(now - t->m_preExitTime) ,
(uint64_t)(now - t->m_exitTime) );
}
// calling thread callback
//log("calling thread id %"INT32" callback", (int32_t)(t->m_tid));
// first call it's callback
//callback ( state );
// clear after just in case
//g_errno = 0;
// debug msg
//log("CLN2 TID=%"UINT32" t=%"INT32"",(int32_t)t->m_tid ,(int32_t)t);
// return now if tt was specified
//if ( tt ) return;
}
int64_t took2 = gettimeofdayInMilliseconds()-startTime;
if(numCallbacks > 0 && took2 > 5)
log(LOG_DEBUG, "threads: took %"INT64" ms to callback %"INT32" "
"callbacks, nice: %"INT32"", took2, numCallbacks, maxNiceness);
//since we need finer grained control in loop, we no longer collect
//the callbacks, sort, then call them. we now call them right away
//that way we can break out if we start taking too long and
//give control back to udpserver.
return numCallbacks != 0;
/*
// print out that we got them
if ( g_conf.m_logDebugThread ) {
int64_t now = gettimeofdayInMilliseconds();
for ( int32_t i = 0 ; i < numCallbacks ; i++ )
log(LOG_DEBUG,"thread: [tid=%"PTRFMT"] %s done3. "
"active=%"INT32" "
"time since queued = %"UINT64" ms "
"time since launch = %"UINT64" ms "
"time since pre-exit = %"UINT64" ms "
"time since exit = %"UINT64" ms",
(PTRTYPE)tids[i],
getThreadType() ,
(int32_t)(m_launched - m_returned) ,
(uint64_t)(now - times [i]),
(uint64_t)(now - times2[i]) ,
(uint64_t)(now - times3[i]) ,
(uint64_t)(now - times4[i]) );
}
// . before calling callbacks, launch any other threads waiting in
// this queue
// . TODO: break into parts, cleanup, launch, call callbacks
//while ( launchThread() );
// . sort callbacks by queued time
// . do bubble sort cuz usually it's not too many threads we cleaned up
bool flag ;
void (* tmpCallback)(void *state,ThreadEntry *t);
int64_t tmpTime;
void *tmpState;
int64_t now = gettimeofdayInMilliseconds();
bubble:
flag = false;
for ( int32_t i = 1 ; i < numCallbacks ; i++ ) {
if ( times[i] >= times[i-1] ) continue;
tmpTime = times [i ];
tmpState = states [i ];
tmpCallback = callbacks[i ];
times [i ] = times [i-1];
states [i ] = states [i-1];
tids [i ] = tids [i-1];
callbacks [i ] = callbacks[i-1];
times [i-1] = tmpTime;
states [i-1] = tmpState;
callbacks [i-1] = tmpCallback;
flag = true;
}
if ( flag ) goto bubble;
// call the callbacks now in order of oldest queued time first
for ( int32_t i = 0 ; i < numCallbacks ; i++ ) {
g_errno = 0;
callbacks[i] ( states[i] , NULL );
}
int64_t took = gettimeofdayInMilliseconds()-now;
if(numCallbacks > 0 && took > 5)
log(LOG_TIMING, "admin: took %"INT64" ms to callback %"INT32" "
"callbacks, nice: %"INT32"", took, numCallbacks, maxNiceness);
#if 0 //I think this is more efficient, for now we'll just use the old way
//theres no reason to sort these, lets just find the top one
//and call it. do it again it until we've called all of them.
int32_t newNumCallbacks = numCallbacks;
for ( int32_t i = 0 ; i < numCallbacks ; i++ ) {
int64_t maxTime = 0;
int32_t maxNdx = 0;
for ( int32_t j = 0 ; j < newNumCallbacks ; j++ ) {
if(maxTime >= times[i]) {
maxTime = times[i];
maxNdx = i;
}
}
g_errno = 0;
callbacks[maxNdx] ( states[maxNdx] );
//copy last one into called slot and decrement
newNumCallbacks--;
times [maxNdx] = times [newNumCallbacks];
states [maxNdx] = states [newNumCallbacks];
callbacks [maxNdx] = callbacks[newNumCallbacks];
}
#endif
g_errno = 0;
// close more threads if we were limited by callbacks[] size
if ( numCallbacks >= 64 ) goto top;
//return true if we called something back;
//returns wrong value if numCallbacks was exactly 64, not too serious...
return numCallbacks != 0;
*/
}
// used by UdpServer to see if it should call a low priority callback
bool Threads::hasHighPriorityCpuThreads() {
ThreadQueue *q ;
//int32_t hiActive = 0 ;
ThreadEntry *t ;
q = &g_threads.m_threadQueues[INTERSECT_THREAD];
t = q->m_launchedHead;
for ( ; t ; t = t->m_nextLink )
if ( t->m_niceness == 0 ) return true;
//hiActive += q->m_hiLaunched - q->m_hiReturned;
q = &g_threads.m_threadQueues[MERGE_THREAD];
t = q->m_launchedHead;
for ( ; t ; t = t->m_nextLink )
if ( t->m_niceness == 0 ) return true;
//hiActive += q->m_hiLaunched - q->m_hiReturned;
return false;
}
// used by UdpServer to see if it should call a low priority callback
int32_t Threads::getNumActiveHighPriorityThreads() {
ThreadQueue *q ;
ThreadEntry *t;
int32_t hiActive = 0 ;
q = &g_threads.m_threadQueues[DISK_THREAD];
t = q->m_launchedHead;
for ( ; t ; t = t->m_nextLink )
if ( t->m_niceness == 0 ) hiActive++;
//hiActive += q->m_hiLaunched - q->m_hiReturned;
q = &g_threads.m_threadQueues[INTERSECT_THREAD];
t = q->m_launchedHead;
for ( ; t ; t = t->m_nextLink )
if ( t->m_niceness == 0 ) hiActive++;
//hiActive += q->m_hiLaunched - q->m_hiReturned;
q = &g_threads.m_threadQueues[MERGE_THREAD];
t = q->m_launchedHead;
for ( ; t ; t = t->m_nextLink )
if ( t->m_niceness == 0 ) hiActive++;
//hiActive += q->m_hiLaunched - q->m_hiReturned;
return hiActive;
}
// . returns false if no thread launched
// . returns true if thread was launched
// . sets g_errno on error
// . don't launch a low priority thread if a high priority thread is running
// . i.e. don't launch a high niceness thread if a low niceness is running
bool ThreadQueue::launchThread2 ( ) {
// or if no stacks left, don't even try
if ( s_head == -1 ) return false;
// . how many threads are active now?
// . NOTE: not perfectly thread safe here
int64_t active = m_launched - m_returned ;
// debug msg
if ( g_conf.m_logDebugThread && m_threadType == DISK_THREAD )
log(LOG_DEBUG,"thread: q=%s launchThread: active=%"INT64" "
"max=%"INT32".",getThreadType(), active,m_maxLaunched);
// return if the max is already launched for this thread queue
if ( active >= m_maxLaunched ) return false;
if ( m_threadType != DISK_THREAD ) {
// if one thread of this type is already out, forget it
// then we can't have 100 GENERIC THREADS!!! with this...
//if ( m_launchedHead ) return false;
// first try niceness 0 queue
ThreadEntry **bestHeadPtr = &m_waitHead0;
ThreadEntry **bestTailPtr = &m_waitTail0;
// if empty, try niceness 1
if ( ! *bestHeadPtr ) {
bestHeadPtr = &m_waitHead1;
bestTailPtr = &m_waitTail1;
}
// then niceness 2
if ( ! *bestHeadPtr ) {
bestHeadPtr = &m_waitHead2;
bestTailPtr = &m_waitTail2;
}
// if bother empty, that was easy
if ( ! *bestHeadPtr ) return false;
// do not launch a low priority merge, intersect or filter
// thread if we/ have high priority cpu threads already going
// on. this way a low priority spider thread will not launch
// if a high priority cpu-based thread of any kind (right now
// just MERGE or INTERSECT) is already running.
if ( (*bestHeadPtr)->m_niceness > 0 &&
g_threads.hasHighPriorityCpuThreads ( ) )
return false;
// otherwise launch the next one
// MERGE_THREAD, INTERSECT, FILTER, UNLINK, SAVETREE
return launchThreadForReals ( bestHeadPtr , bestTailPtr );
}
// disk thread is much more complicated
// use several queues
// debug msg
//log("trying to launch for type=%"INT32"",(int32_t)m_threadType);
// clean up any threads that have exited
//cleanUp ();
// if no entries then nothing to launch
//if ( m_top <= 0 ) return false;
// return log("MAX. %"INT32" are launched. %"INT32" now in queue.",
// active , m_entriesUsed );
// . sanity check
// . a thread can NOT call this
//if ( getpid() != s_pid ) {
// fprintf(stderr,"thread: launchThread: bad engineer\n");
// ::exit(-1);
//}
//int64_t now = gettimeofdayInMilliseconds();
/*
int64_t now = -1LL;
// pick thread with lowest niceness first
int32_t minNiceness = 0x7fffffff;
int64_t maxWait = -1;
//int32_t mini = -1;
//bool minIsWrite = false;
int32_t lowest = 0x7fffffff;
int32_t highest = 0;
// . now base our active thread counts on niceness AND read sizes
// . this is only used for DISK_THREADs
// . loActive* includes niceness >= 1
int32_t loActiveBig = m_loLaunchedBig - m_loReturnedBig;
int32_t loActiveMed = m_loLaunchedMed - m_loReturnedMed;
int32_t loActiveSma = m_loLaunchedSma - m_loReturnedSma;
int32_t mdActiveBig = m_mdLaunchedBig - m_mdReturnedBig;
int32_t mdActiveMed = m_mdLaunchedMed - m_mdReturnedMed;
int32_t mdActiveSma = m_mdLaunchedSma - m_mdReturnedSma;
int32_t hiActiveBig = m_hiLaunchedBig - m_hiReturnedBig;
int32_t hiActiveMed = m_hiLaunchedMed - m_hiReturnedMed;
int32_t hiActiveSma = m_hiLaunchedSma - m_hiReturnedSma;
int32_t activeWrites = m_writesLaunched - m_writesReturned;
// how many niceness=2 threads are currently running now?
int64_t loActive = m_loLaunched - m_loReturned;
int64_t mdActive = m_mdLaunched - m_mdReturned;
//int64_t hiActive = m_hiLaunched - m_hiReturned;
int32_t total = loActive + mdActive;
*/
// hi priority max
// JAB: warning abatement
//int64_t hiActive = m_hiLaunched - m_hiReturned;
// get lowest niceness level of launched threads
ThreadEntry *t = m_launchedHead;
bool hasLowNicenessOut = false;
int32_t activeCount = 0;
int32_t spiderCount = 0;
// int32_t launchedBig0 = 0 , launchedMed0 = 0 , launchedSma0 = 0;
// int32_t launchedBig1 = 0 , launchedMed1 = 0 , launchedSma1 = 0;
for ( ; t ; t = t->m_nextLink ) {
// if he's done, skip him. maybe he hasn't been cleaned up yet
if ( t->m_isDone ) continue;
// count active out
activeCount++;
// get the highest niceness for all that are launched
//if ( niceness > highest ) highest = niceness;
//int32_t rs = t->m_bytesToGo;
// is the launched thread niceness 0? (i.e. high priority)
if ( t->m_niceness == 0 ) {
// set a flag
hasLowNicenessOut = true;
// count read sizes
// if (rs > g_conf.m_medReadSize ) launchedBig0++;
// else if (rs > g_conf.m_smaReadSize ) launchedMed0++;
// else launchedSma0++;
}
else {
spiderCount++;
}
// if ( rs > g_conf.m_medReadSize ) launchedBig1++;
// else if ( rs > g_conf.m_smaReadSize ) launchedMed1++;
// else launchedSma1++;
}
// int32_t max = g_conf.m_spiderMaxDiskThreads;
// if ( max <= 0 ) max = 1;
// if ( activeCount >= max )
// return false;
// get best thread candidate from best linked list of candidates
ThreadEntry **bestHeadPtr = NULL;
ThreadEntry **bestTailPtr = NULL;
// short/med/long high priority (niceness 0) disk reads in head0/1/2
// but we can't launch one more if already at our quota.
if ( ! bestHeadPtr && m_waitHead0 ) {
//launchedSma0 < g_conf.m_queryMaxSmaDiskThreads ) {
bestHeadPtr = &m_waitHead0;
bestTailPtr = &m_waitTail0;
}
// if ( ! bestHeadPtr &&
// m_waitHead1 &&
// launchedMed0 < g_conf.m_queryMaxMedDiskThreads ) {
// bestHeadPtr = &m_waitHead1;
// bestTailPtr = &m_waitTail1;
// }
// if ( ! bestHeadPtr &&
// m_waitHead2 &&
// launchedBig0 < g_conf.m_queryMaxBigDiskThreads ) {
// bestHeadPtr = &m_waitHead2;
// bestTailPtr = &m_waitTail2;
// }
// if we have a niceness 0 disk read/write outstandind and we are
// 1 or 2, do not launch! we do not want low priority disk reads
// having to contend with high priority ones.
// now we only do this if the 'separate disk reads' parms is true.
if ( g_conf.m_separateDiskReads && hasLowNicenessOut && ! bestHeadPtr )
return false;
// threads to save conf and tree/bucket files to disk go in waitHead3
// no these use SAVETREE Thread type
// if ( ! bestHead ) {
// bestHead = m_waitHead3;
// bestTail = m_waitTail3;
// }
// do not allow too high niceness read threads out
if ( spiderCount >= g_conf.m_spiderMaxDiskThreads )
return false;
// low priority (merge or dump) disk WRITES go in waithead4
if ( ! bestHeadPtr && m_waitHead4 ) {
bestHeadPtr = &m_waitHead4;
bestTailPtr = &m_waitTail4;
}
// niceness 1. for merge reads so they superscede regular spider reads
// niceness 1 read threads:
if ( ! bestHeadPtr && m_waitHead5 ) {
bestHeadPtr = &m_waitHead5;
bestTailPtr = &m_waitTail5;
}
// niceness 2 read threads:
if ( ! bestHeadPtr && m_waitHead6 ) {
bestHeadPtr = &m_waitHead6;
bestTailPtr = &m_waitTail6;
}
// if nobody waiting, return false
if ( ! bestHeadPtr ) return false;
// i dunno what the point of this was... so i commented it out
//int32_t max2 = g_conf.m_queryMaxDiskThreads ;
//if ( max2 <= 0 ) max2 = 1;
// only do this check if we're a addlists/instersect thread queue
//if (m_threadType == INTERSECT_THREAD&& hiActive >= max2)return false;
// point to entry in the best linked list to launch from
return launchThreadForReals ( bestHeadPtr , bestTailPtr );
/*
// loop through candidates
for ( int32_t i = 0 ; i < m_top ; i++ ) {
// skip if not occupied
if ( ! m_entries[i].m_isOccupied ) continue;
int32_t niceness = m_entries[i].m_niceness;
// get lowest niceness level of launched threads
if ( m_entries[i].m_isLaunched ) {
// if he's done, skip him
if ( m_entries[i].m_isDone ) continue;
// get the highest niceness for all that are launched
if ( niceness > highest ) highest = niceness;
// get the lowest niceness for all that are launched
if ( niceness < lowest ) lowest = niceness;
// continue now since it's already launched
continue;
}
// . these filters really make it so the spider does not
// impact query response time
//if ( niceness >= 1 && hiActive > 0 ) continue;
// don't consider any lows if one already running
//if ( niceness >= 2 && loActive > 0 ) continue;
// don't consider any lows if a hi already running
//if ( niceness >= 2 && hiActive > 0 ) continue;
// don't consider any mediums if one already running
//if ( niceness == 1 && mdActive > 0 ) continue;
// don't consider any mediums if a hi already running
//if ( niceness == 1 && hiActive > 0 ) continue;
//if ( m_threadType == DISK_THREAD ) {
// if ( niceness >= 1 && hiActive > 0 ) continue;
// if ( niceness >= 2 && loActive >= max ) continue;
// if ( niceness == 1 && mdActive >= max ) continue;
//}
// treat niceness 1 as niceness 2 for ranking purposes
// IFF we're not too backlogged with file merges. i.e.
// IFF we're merging faster than we're dumping.
// only merges and dumps have niceness 1 really.
// spider disk reads are all niceness 2.
// Now Rdb::addList just refuses to add data if we have too
// many unmerged files on disk!
// now we use niceness 1 for "real merges" so those reads take
// priority over spider build reads. (8/14/12)
//if(niceness == 1 ) niceness = 2;
// if he doesn't beat or tie us, skip him
if ( niceness > minNiceness ) continue;
// no more than "max" medium and low priority threads should
// be active/launched at any one time
if ( niceness >= 1 && total >= max ) continue;
// int16_tcut
ThreadEntry *t = &m_entries[i];
// what is this guy's read size?
// the filestate provided could have been
//FileState *fs ;
int32_t readSize = 0 ;
bool isWrite = false;
if ( m_threadType == DISK_THREAD ){//&&m_entries[i].m_state ) {
//fs = (FileState *)m_entries[i].m_state;
readSize = t->m_bytesToGo;
isWrite = t->m_doWrite ;
}
if ( isWrite && activeWrites > g_conf.m_maxWriteThreads )
continue;
if ( m_threadType == MERGE_THREAD ||
m_threadType == INTERSECT_THREAD ||
m_threadType == FILTER_THREAD )
if ( niceness > 0 && hiActive2 > 0 )
continue;
// how many threads can be launched for this readSize/niceness?
if ( niceness >= 1 && m_threadType == DISK_THREAD ) {
if ( readSize > g_conf.m_medReadSize ) {
if ( loActiveBig + mdActiveBig >=
g_conf.m_spiderMaxBigDiskThreads )
continue;
}
else if ( readSize > g_conf.m_smaReadSize ) {
if ( loActiveMed + mdActiveMed >=
g_conf.m_spiderMaxMedDiskThreads )
continue;
}
else if ( loActiveSma + mdActiveSma >=
g_conf.m_spiderMaxSmaDiskThreads )
continue;
}
else if ( niceness < 1 && m_threadType == DISK_THREAD ) {
if ( readSize > g_conf.m_medReadSize ) {
if ( hiActiveBig >=
g_conf.m_queryMaxBigDiskThreads )
continue;
}
else if ( readSize > g_conf.m_smaReadSize ) {
if ( hiActiveMed >=
g_conf.m_queryMaxMedDiskThreads )
continue;
}
else if ( hiActiveSma >=
g_conf.m_queryMaxSmaDiskThreads )
continue;
}
// be lazy with this since it uses a significant amount of cpu
if ( now == -1LL ) now = gettimeofdayInMilliseconds();
// how long has this entry been waiting in the queue to launch?
int64_t waited = now - m_entries[i].m_queuedTime ;
// adjust "waited" if it originally had a niceness of 1
if ( m_entries[i].m_niceness >= 1 ) {
// save threads gain precedence
if ( m_threadType == SAVETREE_THREAD )
waited += 49999999;
else if ( m_threadType == UNLINK_THREAD )
waited += 39999999;
else if ( m_threadType == MERGE_THREAD )
waited += 29999999;
else if ( m_threadType == INTERSECT_THREAD )
waited += 29999999;
else if ( m_threadType == FILTER_THREAD )
waited += 9999999;
// if its a write thread... do it quick
else if ( isWrite )
waited += 19999999;
// . if it has waited more than 500 ms it needs
// to launch now...
// . these values seem to be VERY well tuned on
// my production machines, but they may have to
// be tuned for other machines? TODO.
// . they should auto-tune so when merge is more
// important it should starve the other spider reads
else if ( waited >= 500 )
waited += 9999999;
// it hurts for these guys to wait that int32_t
else waited *= 4;
}
// is real merge?
if ( m_entries[i].m_niceness == 1 )
waited += 19999999;
// watch out for clock skew
if ( waited < 0 ) waited = 0;
// if tied, the lowest time wins
if ( niceness == minNiceness && waited <= maxWait ) continue;
// we got a new winner
mini = i;
minNiceness = niceness;
maxWait = waited;
minIsWrite = isWrite;
}
// if no candidate, bail honorably
if ( mini == -1 ) return false;
// . if we've got an urgent thread (niceness=1) going on, do not allow
// niceness=2 threads to launch
// . actually, don't launch *ANY* if doing an urgent merge even if
// no niceness=1 thread currently launched
// . CAUTION! when titledb is merging it dumps tfndb and if tfndb
// goes urgent,make sure it dumps out with niceness 1, otherwise
// this will freeze us!! since titledb won't be able to continue
// merging until tfndb finishes dumping...
// . Now Rdb::addList just refuses to add data if we have too
// many unmerged files on disk! Let others still read from us,
// just not add to us...
//if ( minNiceness >= 2 && g_numUrgentMerges > 0 ) // && lowest <= 1 )
// return false;
// if we're urgent, don't launch a niceness 2 unless he's waited
// at least 200ms in the queue
//if ( minNiceness >= 2 && g_numUrgentMerges > 0 && maxWait < 200 )
// return false;
// . if the thread to launch has niceness > lowest launched then bail
// . i.e. don't launch a low-priority thread if we have highs running
// . we no longer let a niceness of 1 prevent a niceness of 2 from
// launching, this way we can launch merge threads at a niceness
// of 1 w/o hurting the spidering too much, but still giving the
// merge some preferential treatment over the disk so we don't
// RdbDump data faster than we can finish the merge.
if ( minNiceness > lowest && lowest < 1 ) return false;
// . don't launch a low priority thread if there's a high launched
// from any ThreadQueue
// . hi priority is niceness <= 0, med/lo priority is niceness >= 1
//int64_t hiActive = g_threads.m_hiLaunched - g_threads.m_hiReturned;
// now just limit it to this queue... so you can launch a low
// merge thread even if there's a high disk read going on
//if ( minNiceness >= 1 && hiActive > 0 ) return false;
// if we're low priority and suspended is true then bail
//if ( minNiceness > 0 && m_isLowPrioritySuspended ) return false;
// if we're going to launch a high priority thread then CANCEL
// any low priority disk-read threads already in progress
//if ( minNiceness <= 0 && highest > 0 ) cancelLowPriorityThreads ();
// . let's cancel ALL low priority threads if we're launching a high
// . hi priority is niceness <= 0, low priority is niceness >= 1
//int64_t loActive = g_threads.m_loLaunched - g_threads.m_loReturned;
int32_t realNiceness = m_entries[mini].m_niceness;
//if ( realNiceness <= 0 && (loActive + mdActive) > 0 )
// // this actually cancels medium priority threads, too
// g_threads.cancelLowPriorityThreads();
// point to winning entry
ThreadEntry *t = &m_entries[mini];
*/
}
bool ThreadQueue::launchThreadForReals ( ThreadEntry **headPtr ,
ThreadEntry **tailPtr ) {
ThreadEntry *t = *headPtr;
// if descriptor was closed, just return error now, we
// cannot try to re-open because the file might have been
// unlinked. Sync.cpp does a DISK_THREAD but does not pass in a valid
// FileState ptr because it does its own saving, so check for NULLs.
FileState *fs = (FileState *)t->m_state;
bool allocated = false;
if ( m_threadType == DISK_THREAD && fs && ! fs->m_doWrite ) {
// allocate the read buffer here!
if ( ! fs->m_doWrite && ! fs->m_buf && t->m_bytesToGo > 0 ) {
int32_t need = t->m_bytesToGo + fs->m_allocOff;
char *p = (char *) mmalloc ( need , "ThreadReadBuf" );
if ( p ) {
fs->m_buf = p + fs->m_allocOff;
fs->m_allocBuf = p;
fs->m_allocSize = need;
allocated = true;
}
else
log("thread: read buf alloc failed "
"for %"INT32" bytes.",need);
// just let the BigFile::readWrite_r() handle the
// error for the NULL read buf
}
// . otherwise, they are intact, so get the real fds
// . we know the stored File is still around because of that
bool doWrite = fs->m_doWrite;
BigFile *bb = fs->m_this;
fs->m_fd1 = bb->getfd (fs->m_filenum1,!doWrite);//&fs->m_vfd1);
fs->m_fd2 = bb->getfd (fs->m_filenum2,!doWrite);//&fs->m_vfd2);
// is this bad?
if ( fs->m_fd1 < 0 ) log("disk: fd1 is %i for %s",
fs->m_fd1,bb->getFilename());
if ( fs->m_fd2 < 0 ) log("disk: fd2 is %i for %s.",
fs->m_fd2,bb->getFilename());
fs->m_closeCount1 = getCloseCount_r ( fs->m_fd1 );
fs->m_closeCount2 = getCloseCount_r ( fs->m_fd2 );
}
// count it as launched now, before we actually launch it
m_launched++;
/*
// priority-based GLOBAL & LOCAL launch count
if ( realNiceness <= 0 ) m_hiLaunched++;
else if ( realNiceness == 1 ) m_mdLaunched++;
else if ( realNiceness >= 2 ) m_loLaunched++;
// deal with the tiers for disk threads based on read sizes
if ( m_threadType == DISK_THREAD ) {
// writes are special cases
if ( t->m_doWrite ) m_writesLaunched++;
//FileState *fs = (FileState *)m_entries[mini].m_state;
int32_t rs = t->m_bytesToGo; // 0;
//if ( fs ) rs = fs->m_bytesToGo;
if ( realNiceness >= 2 ) {
if ( rs > g_conf.m_medReadSize )
m_loLaunchedBig++;
else if ( rs > g_conf.m_smaReadSize )
m_loLaunchedMed++;
else
m_loLaunchedSma++;
}
else if ( realNiceness >= 1 ) {
if ( rs > g_conf.m_medReadSize )
m_mdLaunchedBig++;
else if ( rs > g_conf.m_smaReadSize )
m_mdLaunchedMed++;
else
m_mdLaunchedSma++;
}
else {
if ( rs > g_conf.m_medReadSize )
m_hiLaunchedBig++;
else if ( rs > g_conf.m_smaReadSize )
m_hiLaunchedMed++;
else
m_hiLaunchedSma++;
}
}
*/
// debug msg
//if ( m_threadType == 0 )
// log("creating thread, t=%"UINT32" state=%"UINT32"
//launched = %"INT32"",
// t , (int32_t)t->m_state , m_launched );
// and set the flag
t->m_isLaunched = true;
// . launch it
// . this sets the pthread_t ptr for identificatoin
// . returns false on error
//pthread_t tmp;
loop:
// debug msg
if ( g_conf.m_logDebugThread ) {
int32_t active = m_launched - m_returned ;
int64_t now = gettimeofdayInMilliseconds();
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] launched %s thread. "
"active=%"INT32" "
"niceness=%"INT32". waited %"UINT64" ms in queue.",
(PTRTYPE)t, getThreadType(), active, t->m_niceness ,
now - t->m_queuedTime);
}
// be lazy with this since it uses a significant amount of cpu
//if ( now == -1LL ) now = gettimeofdayInMilliseconds();
int64_t now = gettimeofdayInMilliseconds();
//t->m_launchedTime = g_now;
t->m_launchedTime = now;
// loop2:
// spawn the thread
int32_t count = 0;
pid_t pid;
#ifndef PTHREADS
//int status;
//int ret;
// random failure test
//if ( rand() %10 == 1 ) { err = ENOMEM; goto hadError; }
// malloc twice the size
t->m_stackSize = STACK_SIZE;
//t->m_stack = (char *)mmalloc ( t->m_stackSize , "Threads" );
int32_t si = g_threads.getStack ( );
if ( si < 0 ) {
log(LOG_LOGIC,"thread: Unable to get stack. Bad engineer.");
goto hadError;
}
t->m_si = si;
t->m_stack = s_stackPtrs [ si ];
// UNprotect the whole stack so we can use it
mprotect ( t->m_stack + GUARDSIZE , STACK_SIZE - GUARDSIZE ,
PROT_READ | PROT_WRITE );
// clear g_errno
g_errno = 0;
// . make another process
// . do not use sig handlers, so if a child process gets any unhandled
// signal (like SEGV) it will just exit
pid = clone ( startUp , t->m_stack + t->m_stackSize ,
CLONE_FS | CLONE_FILES | CLONE_VM | //CLONE_SIGHAND |
SIGCHLD ,
(void *)t );
// . we set the pid because we are the one that checks it!
// . if we just let him do it, when we check in cleanup routine
// we can get an uninitialized pid
t->m_pid = pid;
// might as well bitch if we should here
if ( s_bad ) {
log(LOG_LOGIC,"thread: PID received: %"INT32" > %"INT32". Bad.",
s_badPid, (int32_t)MAX_PID);
//char *xx = NULL; *xx = 0;
}
// wait for him
//ret = waitpid ( -1*pid , &status , 0 );
//if ( ret != pid )
// log("waitpid(pid=%"INT32"): ret=%"INT32" err=%s",
// (int32_t)pid,(int32_t)ret,mstrerror(errno));
// check if he's done
//if ( ! t->m_isDone ) log("NOT DONE");
// set the pid
//t->m_pid = pid;
// error?
if ( pid == (pid_t)-1 ) g_errno = errno;
//
// now use pthreads again... are they stable yet?
//
#else
// assume it does not go through
t->m_needsJoin = false;
// pthread inherits our sigmask, so don't let it handle sigalrm
// signals in Loop.cpp, it'll screw things up. that handler
// is only meant to be called by the main process. if we end up
// double calling it, this thread may think g_callback is non-null
// then it gets set to NULL, then the thread cores! seen it...
// sigset_t sigs;
// sigemptyset ( &sigs );
// sigaddset ( &sigs , SIGALRM );
// sigaddset ( &sigs , SIGVTALRM );
// if ( sigprocmask ( SIG_BLOCK , &sigs , NULL ) < 0 )
// log("threads: failed to block sig");
// supply our own stack to make pthread_create() fast otherwise
// it has slowness issues with mmap()
// http://www.gossamer-threads.com/lists/linux/kernel/960227
t->m_stackSize = STACK_SIZE;
//t->m_stack = (char *)mmalloc ( t->m_stackSize , "Threads" );
int32_t si = g_threads.getStack ( );
if ( si < 0 ) {
log(LOG_LOGIC,"thread: Unable to get stack. Bad engineer.");
goto hadError;
}
t->m_si = si;
t->m_stack = s_stackPtrs [ si ];
// check it's aligned
if ( (uint64_t)(t->m_stack) & (THRPAGESIZE-1) ) {
char *xx=NULL;*xx=0; }
// UNprotect the whole stack so we can use it
mprotect ( t->m_stack + GUARDSIZE , STACK_SIZE - GUARDSIZE ,
PROT_READ | PROT_WRITE );
pthread_attr_t attr;
pthread_attr_init ( &attr );
pthread_attr_setstack ( &attr , t->m_stack , t->m_stackSize );
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"remove from wait list, add to launch list",
(PTRTYPE)t);
// remove from waiting linked list, whichever one it was in
removeLink2 ( headPtr , tailPtr , t );
// add to 'launched' linked list
addLink ( &m_launchedHead , t );
// save the waiting linked list we came from in case we get cancelled
// and we have to put it back into it
t->m_bestHeadPtr = headPtr;
t->m_bestTailPtr = tailPtr;
// debug
if ( g_conf.m_logDebugThread )
log("threads: pthread_create: "
"stack=%"PTRFMT" stacksize=%"INT64""
, (PTRTYPE)t->m_stack
, (int64_t)t->m_stackSize );
// this returns 0 on success, or the errno otherwise
g_errno = pthread_create ( &t->m_joinTid , &attr, startUp2 , t) ;
// if ( sigprocmask ( SIG_UNBLOCK , &sigs , NULL ) < 0 )
// log("threads: failed to unblock sig");
#endif
// we're back from pthread_create
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: Back from clone "
"t=0x%"PTRFMT" pid=%"INT32".",
(PTRTYPE)t,(int32_t)pid);
// return true on successful creation of the thread
if ( g_errno == 0 ) {
// good stuff, the thread needs a join now
t->m_needsJoin = true;
if ( count > 0 )
log("thread: Call to clone looped %"INT32" times.",
count);
return true;
}
//#undef usleep
// forever loop
if ( g_errno == EAGAIN ) {
if ( count++ == 0 )
log("thread: Call to clone had error: %s.",
mstrerror(g_errno));
//usleep(1);
//goto loop2;
goto hadError;
}
// debug msg
// log("created tid=%"INT32"",(int32_t)t->m_tid);
// return true;
//}
// do again if g_errno is AGAIN
if ( g_errno == EINTR ) {
log("thread: Call to clone was interrupted. Trying again.");
goto loop;
}
//#ifndef _PTHREADS_
hadError:
//#endif
if ( g_errno )
log("thread: pthread_create had error = %s",
mstrerror(g_errno));
// it didn't launch, did it? dec the count.
m_launched--;
// re-protect this stack
mprotect ( t->m_stack + GUARDSIZE , STACK_SIZE - GUARDSIZE,
PROT_NONE );
// RETURN THE STACK
g_threads.returnStack ( t->m_si );
t->m_stack = NULL;
/*
// priority-based LOCAL & GLOBAL launch counts
if ( realNiceness <= 0 ) m_hiLaunched--;
else if ( realNiceness == 1 ) m_mdLaunched--;
else if ( realNiceness >= 2 ) m_loLaunched--;
// . deal with the tiers for disk threads based on read sizes
// . WARNING: we cannot easily change tiers dynamically
// because it will throw these counts off
if ( m_threadType == DISK_THREAD ) {
// writes are special cases
if ( t->m_doWrite ) m_writesLaunched--;
//FileState *fs = (FileState *)m_entries[mini].m_state;
int32_t rs = t->m_bytesToGo; // 0;
//if ( fs ) rs = fs->m_bytesToGo;
if ( realNiceness >= 2 ) {
if ( rs > g_conf.m_medReadSize )
m_loLaunchedBig--;
else if ( rs > g_conf.m_smaReadSize )
m_loLaunchedMed--;
else
m_loLaunchedSma--;
}
else if ( realNiceness >= 1 ) {
if ( rs > g_conf.m_medReadSize )
m_mdLaunchedBig--;
else if ( rs > g_conf.m_smaReadSize )
m_mdLaunchedMed--;
else
m_mdLaunchedSma--;
}
else {
if ( rs > g_conf.m_medReadSize )
m_hiLaunchedBig--;
else if ( rs > g_conf.m_smaReadSize )
m_hiLaunchedMed--;
else
m_hiLaunchedSma--;
}
}
*/
// unset the flag
t->m_isLaunched = false;
// bail on other errors
log("thread: Call to clone had error: %s.", mstrerror(g_errno));
// correction on this error
log("thread: Try not using so much memory. "
"memused now =%"INT64".",g_mem.getUsedMem());
// free allocated buffer
if ( allocated ) {
mfree ( fs->m_allocBuf , fs->m_allocSize , "ThreadReadBuf" );
fs->m_buf = NULL;
}
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"remove from launched list, RE-add to wait list",
(PTRTYPE)t);
// remove from launched linked list
removeLink ( &m_launchedHead , t );
// back into the queue waiting to launch
addLinkToHead ( headPtr , tailPtr , t );
// i'm not sure return value matters at this point? the thread
// is queued and hopefully will launch at some point
return false;
/*
// if this is the direct thread request do not call callback, just
// return false, otherwise we get into an unexpected loop thingy
if ( t == te )
return log("thread: Returning false.");
// do it blocking
log("thread: Calling without thread. This will crash many times. "
"Please fix it.");
// return false so caller will re-do without thread!
// so BigFile::readwrite() will retry without thread and we won't
// get into a weird loop thingy
if ( te ) return false;
// uint64_t profilerStart,profilerEnd;
// uint64_t statStart,statEnd;
//if (g_conf.m_profilingEnabled){
// address=(int32_t)t->m_startRoutine;
// g_profiler.startTimer(address, __PRETTY_FUNCTION__);
//}
t->m_startRoutine ( t->m_state , t );
//if (g_conf.m_profilingEnabled) {
// if(!g_profiler.endTimer(address, __PRETTY_FUNCTION__))
// log(LOG_WARN,"admin: Couldn't add the fn %"INT32"",
// (int32_t)address);
//}
t->m_exitTime = gettimeofdayInMilliseconds();
// flag it for cleanup
t->m_isDone = true;
t->m_isLaunched = true;
// clean it up
cleanUp ( t , 200);//maxNiceness thread can have to be cleaned up
// ignore error
g_errno = 0;
// we kinda launched one, so say true here
return true; // false;
*/
}
#ifndef PTHREADS
static bool s_firstTime = true;
#endif
// threads start up with cacnellation deferred until pthreads_testcancel()
// is called, but we never call that
int startUp ( void *state ) {
// get thread entry
ThreadEntry *t = (ThreadEntry *)state;
// no! now parent does since he is the one that needs to check it
// in the cleanup routine
// remember the pid
//t->m_pid = getpid();
// . sanity check
// . a thread can NOT call this
#ifndef PTHREADS
if ( getpid() == s_pid )
log("thread: Thread has same pid %i as main process.",s_pid);
#endif
// the cleanup handler
//pthread_cleanup_push ( exitWrapper , t ) ; // t->m_state );
// our signal set
sigset_t set;
sigemptyset(&set);
//sigaddset(&set, SIGHUP);
// we need this here so if we break the gb process with gdb it
// does not kill the child processes when it sends out the SIGINT.
sigaddset(&set, SIGINT);
// ignore the real time signal, man...
//sigaddset(&set, GB_SIGRTMIN);
//pthread_sigmask(SIG_BLOCK, &set, NULL);
#ifndef PTHREADS
sigprocmask(SIG_BLOCK, &set, NULL);
#else
// turn these off in the thread
sigaddset ( &set , SIGALRM );
sigaddset ( &set , SIGVTALRM );
pthread_sigmask(SIG_BLOCK,&set,NULL);
#endif
// . what this lwp's priority be?
// . can range from -20 to +20
// . the lower p, the more cpu time it gets
// . this is really the niceness, not the priority
int p ;
// currently our niceness ranges from -1 to 2 for us
if ( t->m_niceness == 2 ) p = 19 ;
else if ( t->m_niceness == 1 ) p = 10 ;
else p = 0 ;
// remember the tid
//t->m_tid = pthread_self();
// debug
if ( g_conf.m_logDebugThread )
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"in startup pid=%"INT64" pppid=%"INT32"",
(PTRTYPE)t,(int64_t)getpidtid(),(int32_t)getppid());
// debug msg
//fprintf(stderr,"new thread tid=%"INT32" pid=%"INT32"\n",
// (int32_t)t->m_tid,(int32_t)t->m_pid);
// . set this process's priority
// . setpriority() is only used for SCHED_OTHER threads
//if ( pthread_setschedprio ( getpidtid() , p ) ) {
#ifndef PTHREADS
if ( setpriority ( PRIO_PROCESS, getpidtid() , p ) < 0 ) {
// do we even support logging from a thread?
if ( s_firstTime ) {
log("thread: Call to "
"setpriority(%"UINT32",%i) in thread "
"failed: %s. This "
"message will not be repeated.",
(uint32_t)getpidtid(),p,mstrerror(errno));
s_firstTime = false;
}
errno = 0;
}
#endif
/*
sched_param sp;
int pp;
int err = pthread_getschedparam ( t->m_tid , &pp , &sp ) ;
if ( err ) log("thread: startUp: pthread_getschedparam: %s",
mstrerror(err));
// adjust the priority
p = 1;
sp.sched_priority = p;
// and reassign
err = pthread_setschedparam ( t->m_tid , SCHED_OTHER , &sp ) ;
log("thread: startUp: pthread_setschedparam(%"INT32"): %s",
p,mstrerror(err));
*/
// somehow, it works ok when we have this print statement delay!!!
//fprintf(stderr,"thread pid = %"INT32"\n",(int32_t)getpid());
// . call the startRoutine
// . IMPORTANT: this can NEVER do non-blocking stuff
t->m_startRoutine ( t->m_state , t );
// pop it off
//pthread_cleanup_pop ( 1 /*execute handler?*/ );
// . now throw it on g_loop's sigqueue
// . the first 4 bytes of t->m_state should be t->m_callback
// . no! just use 1 to tell Loop to call g_threads.cleanUp()
// . TODO: pass in a ptr to cleanUpWrapper() instead of "t"
// . sival_int is only 4 bytes on a 64 bit arch...
sigval_t svt;
svt.sival_int = 1;//(int64_t)t ; //(int)(t->m_state); // fd;
// set exit time
int64_t now = gettimeofdayInMilliseconds();
t->m_preExitTime = now;
t->m_exitTime = now;
if ( g_conf.m_logDebugThread ) {
log(LOG_DEBUG,"thread: [t=0x%"PTRFMT"] "
"done with startup pid=%"INT64"",
(PTRTYPE)t,(int64_t)getpidtid());
}
// . now mark thread as ready for removal
// . do this BEFORE queueing the signal since we're still a thread!!!
// . cleanUp() will take care of the rest
// . cleanUp() will call pthread_join on us!
t->m_isDone = true;
// let Loop.cpp's sigHandler_r call g_thread.cleanUp()
g_threads.m_needsCleanup = true;
//if(t->m_niceness > 0) g_threads.m_needBottom = true;
// . send the signal
// . if queue is full g_loop will get a SIGIO and call
// g_threads.cleanUp()/launchThreads() in it's doPoll() routine
// . we reserve GB_SIGRTMIN itself for unblocked interrupts for
// UdpServer
// . damn, it seems that if the queue is full pthread_join is unable
// to detach threads.... so sleep until it clears up
// . HEY! process is supposed to send us an ECHLD signal? right?
//sigqueue ( s_pid, GB_SIGRTMIN + 1 + t->m_niceness, svt ) ;
// . it does not send us a signal automatically, so we must do it!
// . i noticed during the linkdb rebuild we were not getting the signal
//sigqueue ( s_pid, GB_SIGRTMIN + 1 + t->m_niceness, svt ) ;
// i verified this breaks select() in Loop.cpp out of it's sleep
//fprintf(stderr,"threads sending SIGCHLD\n");
// try a sigchld now! doesn't it already do this? no...
// on 64bit arch pthread_t is 64bit and pid_t is 32bit
// i dont think this makes sense with pthreads any more, they don't
// use pid_t they use pthread_t
#ifndef PTHREADS
sigqueue ( (pid_t)(int64_t)s_pid, SIGCHLD, svt ) ;
#else
pthread_kill ( s_pid , SIGCHLD );
#endif
return 0;
}
// pthread_create uses this one
void *startUp2 ( void *state ) {
startUp ( state );
return NULL;
}
// watch out, UdpServer::getEmptySlot() calls UdpServer::suspend() and we
// could be in a signal handler here
void ThreadQueue::suspendLowPriorityThreads() {
// disable for now
//return;
// just return if already suspended
if ( m_isLowPrioritySuspended ) return;
// log it
//log("ThreadQueue:: suspending low priority threads!!!!!!");
// set the flag so low priority threads won't be launched
m_isLowPrioritySuspended = true;
// . cancel any outstanding low priority threads that are running
// . no, only cancel if we run another disk thread
// . this will happen above in ThreadQueue::launchThread()
//cancelLowPriorityThreads();
}
// this is called by UdpServer::destroySlot() and should NOT be in a sig handlr
void ThreadQueue::resumeLowPriorityThreads() {
// disable for now
//return;
// bail if no need
if ( ! m_isLowPrioritySuspended ) return;
// turn em back on
m_isLowPrioritySuspended = false;
// just in case, though
//if ( g_inSigHandler ) {
// log(LOG_LOGIC,"thread: resumeLowPriorityThreads: In sig "
// "handler.");
// return;
//}
// try to start up some threads then
g_threads.launchThreads();
}
void ThreadQueue::print ( ) {
// loop through candidates
for ( int32_t i = 0 ; i < m_top ; i++ ) {
ThreadEntry *t = &m_entries[i];
// print it
log(LOG_INIT,"thread: address=%"PTRFMT" "
"pid=%u state=%"PTRFMT" "
"occ=%i done=%i lnch=%i",
(PTRTYPE)t , t->m_pid ,
(PTRTYPE)t->m_state , t->m_isOccupied , t->m_isDone ,
t->m_isLaunched );
}
}
const char *ThreadQueue::getThreadType ( ) {
const char *s = "unknown";
if ( m_threadType == DISK_THREAD ) s = "disk";
if ( m_threadType == MERGE_THREAD ) s = "merge";
if ( m_threadType == INTERSECT_THREAD ) s = "intersectlists";
if ( m_threadType == FILTER_THREAD ) s = "filter";
if ( m_threadType == SAVETREE_THREAD ) s = "savetree";
if ( m_threadType == UNLINK_THREAD ) s = "unlink";
if ( m_threadType == GENERIC_THREAD ) s = "generic";
return s;
}
#include "BigFile.h" // FileState class
/*
MDW: this is unused
int32_t Threads::getDiskThreadLoad ( int32_t maxNiceness , int32_t *totalToRead ) {
ThreadQueue *q = &m_threadQueues[DISK_THREAD];
ThreadEntry *e = q->m_entries;
int32_t top = q->m_top;
*totalToRead = 0;
int32_t n = 0;
// we really can't suspend threads cuz they might have the
// mutex lock so we just cancel the disk threads here then
for ( int32_t i = 0 ; i < top ; i++ ) {
// get entry
ThreadEntry *t = &e[i];
// skip if not occupied
if ( ! t->m_isOccupied ) continue;
// skip if it's nicer than what we want
if (t->m_niceness > maxNiceness && ! t->m_isLaunched) continue;
// skip if already done
if ( t->m_isDone ) continue;
// cast state data
FileState *fs = (FileState *) t->m_state;
// sometimes NULL, like from Sync.cpp's call
if ( ! fs ) continue;
// only remove read operations, since write operations get
// the fd up front
if ( t->m_doWrite ) continue;
// how many byte to do
//int32_t todo = fs->m_bytesToGo - fs->m_bytesDone;
int32_t todo = t->m_bytesToGo;
// multiply by 2 if a write
if ( t->m_doWrite ) todo *= 2;
// add to total bytes to read
*totalToRead += todo;
// count the thread
n++;
}
return n;
}
*/
// when a BigFile is removed, much like we remove its pages from DiskPageCache
// we also remove any unlaunched reads/writes on it from the thread queue.
void ThreadQueue::removeThreads ( BigFile *bf ) {
// did the BigFile get hosed? that means our BigFile was
// unlinked or closed before we got a chance to launch the
// thread.
//int32_t maxi = -1;
//ThreadEntry *head ;
removeThreads2 ( &m_waitHead0 , &m_waitTail0 , bf );
removeThreads2 ( &m_waitHead1 , &m_waitTail1 , bf );
removeThreads2 ( &m_waitHead2 , &m_waitTail2 , bf );
removeThreads2 ( &m_waitHead3 , &m_waitTail3 , bf );
removeThreads2 ( &m_waitHead4 , &m_waitTail4 , bf );
removeThreads2 ( &m_waitHead5 , &m_waitTail5 , bf );
removeThreads2 ( &m_waitHead6 , &m_waitTail6 , bf );
}
void ThreadQueue::removeThreads2 ( ThreadEntry **headPtr ,
ThreadEntry **tailPtr ,
BigFile *bf ) {
int32_t saved = g_errno;
ThreadEntry *t = *headPtr;
ThreadEntry *nextLink = NULL;
for ( ; t ; t = nextLink ) {
// do it here in case we modify the linked list below
nextLink = t->m_nextLink;
// get the filestate
FileState *fs = (FileState *)t->m_state;
// skip if NULL
if ( ! fs ) continue;
// skip if not match
if ( fs->m_this != (void *)bf ) continue;
// . let it finish writing if it is a write thread
// . otherwise, if we are exiting, we could free the
// buffer being written and cause the thread to core...
if ( fs->m_doWrite ) {
log(LOG_INFO,"disk: Not removing write thread.");
continue;
}
// . should we really? if we renamed the file to another,
// we need to recompute the offsets to read, etc.. so we
// should fail up to Msg5 with EFILECLOSED or something...
// . i think we did a rename and it got the same fd, and since
// we did not remove the launched or done threads after the
// rename, we're not sure if they read from the newly renamed
// file or not, and our read offset was for the old file...
// . at least set the error flag for doneWrapper()
fs->m_errno2 = EFILECLOSED;
// log it
logf(LOG_INFO,"disk: Removing/flagging operation in thread "
"queue. fs=0x%"PTRFMT"", (PTRTYPE)fs);
// skip if already done
if ( t->m_isDone ) continue;
// skip if launched
if ( t->m_isLaunched ) continue;
// note in the log it is launched
log(LOG_INFO,"disk: Thread is launched.");
// tell donewrapper what happened
fs->m_errno = EFILECLOSED;
g_errno = EFILECLOSED;
// note it
//log(LOG_INFO,"disk: Removing operation from thread queue.");
// remove it from the thread queue
t->m_isDone = true;
t->m_isLaunched = false;
t->m_isOccupied = false;
// keep track
//maxi = i;
// remove from waiting linked list, whichever one it was in
removeLink2 ( headPtr , tailPtr , t );
// add to 'empty' linked list
addLink ( &m_emptyHead , t );
makeCallback ( t );
}
// do we have to decrement top
// if ( m_top == maxi + 1 )
// while ( m_top>0 && !m_entries[m_top-1].m_isOccupied) m_top--;
// this was causing us to lose a g_errno value when XmlDoc::~XmlDoc()
// called BigFile::~BigFile() called removeThreads()
//called removeThreads2()
//g_errno = 0;
g_errno = saved;
}
void Threads::printState() {
int64_t now = gettimeofdayInMilliseconds();
for ( int32_t i = 0 ; i < m_numQueues; i++ ) {
ThreadQueue *q = &m_threadQueues[i];
// int32_t loActive = q->m_loLaunched - q->m_loReturned;
// int32_t mdActive = q->m_mdLaunched - q->m_mdReturned;
// int32_t hiActive = q->m_hiLaunched - q->m_hiReturned;
// int32_t total = loActive + mdActive + hiActive;
//if( total == 0) continue;
// log(LOG_TIMING,
// "admin: Thread counts: type:%s "
// "%"INT32":low %"INT32":med%"INT32":high %"INT32":total",
// q->getThreadType(),loActive, mdActive, hiActive, total);
for ( int32_t j = 0 ; j < q->m_top ; j++ ) {
ThreadEntry *t = &q->m_entries[j];
if(!t->m_isOccupied) continue;
if(t->m_isDone) {
log(LOG_TIMING,
"admin: Thread -done- "
"nice: %"INT32" "
"totalTime: %"INT64" (ms) "
"queuedTime: %"INT64"(ms) "
"runTime: %"INT64"(ms) "
"cleanup: %"INT64"(ms) "
"callback:%s",
t->m_niceness,
now - t->m_queuedTime,
t->m_launchedTime - t->m_queuedTime,
t->m_exitTime - t->m_launchedTime,
now - t->m_exitTime,
g_profiler.
getFnName((PTRTYPE)t->m_callback));
continue;
}
if(t->m_isLaunched) {
log(LOG_TIMING,
"admin: Thread -launched- "
"nice: %"INT32" "
"totalTime: %"INT64"(ms) "
"queuedTime: %"INT64"(ms) "
"runTime: %"INT64"(ms) "
"callback:%s",
t->m_niceness,
now - t->m_queuedTime,
t->m_launchedTime - t->m_queuedTime,
now - t->m_launchedTime,
g_profiler.
getFnName((PTRTYPE)t->m_callback));
continue;
}
log(LOG_TIMING,
"admin: Thread -queued- "
"nice: %"INT32" "
"queueTime: %"INT64"(ms) "
"callback:%s",
t->m_niceness,
now - t->m_queuedTime,
g_profiler.getFnName((PTRTYPE)t->m_callback));
}
}
}
void ThreadQueue::killAllThreads ( ) {
for ( int32_t i = 0 ; i < m_maxEntries ; i++ ) {
ThreadEntry *e = &m_entries[i];
if ( ! e->m_isOccupied ) continue;
if ( ! e->m_isLaunched ) continue;
log("threads: killling thread id %i",(int)e->m_joinTid);
pthread_kill ( e->m_joinTid , SIGKILL );
log("threads: joining with thread id %i",(int)e->m_joinTid);
pthread_join ( e->m_joinTid , NULL );
}
}
void Threads::killAllThreads ( ) {
log("threads: killing all threads");
for ( int32_t j = 0 ; j < m_numQueues ; j++ ) {
ThreadQueue *tq = &m_threadQueues[j];
tq->killAllThreads();
}
}
Reference in New Issue View Git Blame Copy Permalink