Mercurial > hg > webserver / file comparison

comparison: src/server/util/thrpool.c

src/server/util/thrpool.c

changeset 738
677bff22d1b2
parent 724
dcac069c7c95
equal deleted inserted replaced
737:b240ff52cee5 738:677bff22d1b2
44 pool->num_idle = 0; 44 pool->num_idle = 0;
45 pool->min_threads = min; 45 pool->min_threads = min;
46 pool->max_threads = max; 46 pool->max_threads = max;
47 pool->num_threads = 0; 47 pool->num_threads = 0;
48 pool->last_job = 0; 48 pool->last_job = 0;
49 pool->last_thread = -1;
50 49
51 pool->threads = calloc(max, sizeof(pthread_t)); 50 pool->threads = calloc(max, sizeof(pthread_t));
52 pool->thrstatus = calloc(max, sizeof(int)); 51 pool->thrstatus = calloc(max, sizeof(int));
53 52
54 pthread_mutex_init(&pool->queue_lock, NULL); 53 pthread_mutex_init(&pool->queue_lock, NULL);
74 73
75 void* threadpool_func(void *data) { 74 void* threadpool_func(void *data) {
76 threadpool_t *pool = (threadpool_t*)data; 75 threadpool_t *pool = (threadpool_t*)data;
77 76
78 pthread_t thr_self = pthread_self(); 77 pthread_t thr_self = pthread_self();
79 int thr_index = -1;
80 for(int i=0;i<pool->max_threads;i++) {
81 if(pool->threads[i] == thr_self) {
82 thr_index = i; // TODO: this is stupid, trasfer the thread index per data
83 break;
84 }
85 }
86
87 if(thr_index == -1) {
88 log_ereport(LOG_CATASTROPHE, "threadpool: cannot find thread index for thread %ull", (unsigned long long)thr_self);
89 return NULL;
90 }
91 78
92 ws_atomic_inc32(&pool->num_threads); 79 ws_atomic_inc32(&pool->num_threads);
93 for(;;) { 80 for(;;) {
94 threadpool_job *job = threadpool_get_job(pool, thr_index); 81 threadpool_job *job = threadpool_get_job(pool, thr_self);
95 if(job == &kill_job) { 82 if(job == &kill_job) {
96 break; 83 break;
97 } 84 }
98 85
99 job->callback(job->data); 86 job->callback(job->data);
105 log_ereport(LOG_INFORM, "threadpool closed"); // TODO: log threadpool name 92 log_ereport(LOG_INFORM, "threadpool closed"); // TODO: log threadpool name
106 } 93 }
107 return NULL; 94 return NULL;
108 } 95 }
109 96
110 threadpool_job* threadpool_get_job(threadpool_t *pool, int thread_index) { 97 threadpool_job* threadpool_get_job(threadpool_t *pool, pthread_t thr) {
111 struct timespec timeout; 98 struct timespec timeout;
112 clock_gettime(CLOCK_REALTIME, &timeout); 99 clock_gettime(CLOCK_REALTIME, &timeout);
113 timeout.tv_sec += 30; 100 timeout.tv_sec += 30;
114 101
115 while(pthread_mutex_timedlock(&pool->queue_lock, &timeout)) { 102 while(pthread_mutex_timedlock(&pool->queue_lock, &timeout)) {
121 pool->num_idle++; 108 pool->num_idle++;
122 while(job == NULL) { 109 while(job == NULL) {
123 if(pool->queue_len == 0) { 110 if(pool->queue_len == 0) {
124 timeout.tv_sec += 30; 111 timeout.tv_sec += 30;
125 while(pthread_cond_timedwait(&pool->available, &pool->queue_lock, &timeout)) { 112 while(pthread_cond_timedwait(&pool->available, &pool->queue_lock, &timeout)) {
126 log_ereport(LOG_DEBUG, "threadpool_get_job: cond timeout: thread: %d queue: %u", thread_index, (unsigned int)pool->queue_len); 113 log_ereport(LOG_DEBUG, "threadpool_get_job: cond timeout: thread: %d queue: %ull", (unsigned long long)thr, (unsigned int)pool->queue_len);
127 timeout.tv_sec += 60; 114 timeout.tv_sec += 60;
128 } 115 }
129 continue; 116 continue;
130 } else { 117 } else {
131 pool_queue_t *q = pool->queue; 118 pool_queue_t *q = pool->queue;
135 free(q); 122 free(q);
136 } 123 }
137 } 124 }
138 pool->num_idle--; 125 pool->num_idle--;
139 126
140 pool->last_thread = thread_index;
141 pool->last_job = time(NULL); 127 pool->last_job = time(NULL);
142 128
143 pthread_mutex_unlock(&pool->queue_lock); 129 pthread_mutex_unlock(&pool->queue_lock);
144 return job; 130 return job;
145 } 131 }

Mercurial Repository: webserver

mercurial