1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 8 -*- */
5 * Copyright (C) 2008, Eduardo Silva P.
7 * This program is free software; you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation; either version 2 of the License, or
10 * (at your option) any later version.
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU Library General Public License for more details.
17 * You should have received a copy of the GNU General Public License
18 * along with this program; if not, write to the Free Software
19 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
26 #include <sys/epoll.h>
28 #include <sys/syscall.h>
32 #include "conn_switch.h"
34 #include "scheduler.h"
42 /* Register thread information */
43 int mk_sched_register_thread(pthread_t tid
, int efd
)
46 struct sched_list_node
*sr
, *aux
;
48 sr
= mk_mem_malloc_z(sizeof(struct sched_list_node
));
52 sr
->queue
= mk_mem_malloc_z(sizeof(struct sched_connection
)*
53 config
->worker_capacity
);
54 sr
->request_handler
= NULL
;
57 for(i
=0; i
<config
->worker_capacity
; i
++){
58 sr
->queue
[i
].status
= MK_SCHEDULER_CONN_AVAILABLE
;
73 sr
->idx
= aux
->idx
+ 1;
80 * Create thread which will be listening
81 * for incomings file descriptors
83 int mk_sched_launch_thread(int max_events
)
88 sched_thread_conf
*thconf
;
89 pthread_mutex_t mutex_wait_register
;
91 /* Creating epoll file descriptor */
92 efd
= mk_epoll_create(max_events
);
99 pthread_mutex_init(&mutex_wait_register
,(pthread_mutexattr_t
*) NULL
);
100 pthread_mutex_lock(&mutex_wait_register
);
102 thconf
= mk_mem_malloc(sizeof(sched_thread_conf
));
103 thconf
->epoll_fd
= efd
;
104 thconf
->max_events
= max_events
;
106 pthread_attr_init(&attr
);
107 pthread_attr_setdetachstate(&attr
, PTHREAD_CREATE_DETACHED
);
108 if(pthread_create(&tid
, &attr
, mk_sched_launch_epoll_loop
,
111 perror("pthread_create");
115 /* Register working thread */
116 mk_sched_register_thread(tid
, efd
);
117 pthread_mutex_unlock(&mutex_wait_register
);
121 /* created thread, all this calls are in the thread context */
122 void *mk_sched_launch_epoll_loop(void *thread_conf
)
124 sched_thread_conf
*thconf
;
125 struct sched_list_node
*thinfo
;
127 /* Avoid SIGPIPE signals */
128 mk_signal_thread_sigpipe_safe();
130 thconf
= thread_conf
;
132 /* Init specific thread cache */
133 mk_cache_thread_init();
135 mk_epoll_calls
*callers
;
136 callers
= mk_epoll_set_callers((void *)mk_conn_switch
,
138 MK_CONN_SWITCH_WRITE
);
140 /* Nasty way to export task id */
142 thinfo
= mk_sched_get_thread_conf();
144 thinfo
= mk_sched_get_thread_conf();
147 /* Glibc doesn't export to user space the gettid() syscall */
148 thinfo
->pid
= syscall(__NR_gettid
);
150 mk_sched_set_thread_poll(thconf
->epoll_fd
);
151 mk_epoll_init(thconf
->epoll_fd
, callers
, thconf
->max_events
);
156 struct request_idx
*mk_sched_get_request_index()
158 return (struct request_idx
*) pthread_getspecific(request_index
);
161 void mk_sched_set_request_index(struct request_idx
*ri
)
163 pthread_setspecific(request_index
, (void *)ri
);
166 void mk_sched_set_thread_poll(int epoll
)
168 pthread_setspecific(epoll_fd
, (void *) epoll
);
171 int mk_sched_get_thread_poll()
173 return (int) pthread_getspecific(epoll_fd
);
176 struct sched_list_node
*mk_sched_get_thread_conf()
178 struct sched_list_node
*node
;
181 current
= pthread_self();
184 if(pthread_equal(node
->tid
, current
) != 0){
194 void mk_sched_update_thread_status(int active
, int closed
)
196 struct sched_list_node
*thnode
;
198 thnode
= mk_sched_get_thread_conf();
201 case MK_SCHEDULER_ACTIVE_UP
:
202 thnode
->active_requests
++;
204 case MK_SCHEDULER_ACTIVE_DOWN
:
205 thnode
->active_requests
--;
210 case MK_SCHEDULER_CLOSED_UP
:
211 thnode
->closed_requests
++;
213 case MK_SCHEDULER_CLOSED_DOWN
:
214 thnode
->closed_requests
--;
219 int mk_sched_add_client(struct sched_list_node
**sched
, int remote_fd
)
222 struct sched_list_node
*l
;
224 l
= (struct sched_list_node
*) *sched
;
226 /* Look for an available slot */
227 for(i
=0; i
<config
->worker_capacity
; i
++){
228 if(l
->queue
[i
].status
== MK_SCHEDULER_CONN_AVAILABLE
){
229 l
->queue
[i
].socket
= remote_fd
;
230 l
->queue
[i
].status
= MK_SCHEDULER_CONN_PENDING
;
231 l
->queue
[i
].arrive_time
= log_current_utime
;
233 mk_epoll_add_client(l
->epoll_fd
, remote_fd
,
234 MK_EPOLL_BEHAVIOR_TRIGGERED
);
242 int mk_sched_remove_client(struct sched_list_node
**sched
, int remote_fd
)
244 struct sched_connection
*sc
;
246 sc
= mk_sched_get_connection(sched
, remote_fd
);
249 sc
->status
= MK_SCHEDULER_CONN_AVAILABLE
;
256 struct sched_connection
*mk_sched_get_connection(struct sched_list_node
**sched
,
260 struct sched_list_node
*l
;
263 l
= mk_sched_get_thread_conf();
266 l
= (struct sched_list_node
*) *sched
;
269 for(i
=0; i
<config
->worker_capacity
; i
++){
270 if(l
->queue
[i
].socket
== remote_fd
){
278 int mk_sched_check_timeouts(struct sched_list_node
**sched
)
281 struct request_idx
*req_idx
;
282 struct client_request
*req_cl
;
283 struct sched_list_node
*l
;
285 l
= (struct sched_list_node
*) *sched
;
287 /* PENDING CONN TIMEOUT */
288 for(i
=0; i
<config
->worker_capacity
; i
++){
289 if(l
->queue
[i
].status
== MK_SCHEDULER_CONN_PENDING
){
290 if(l
->queue
[i
].arrive_time
+ config
->timeout
<= log_current_utime
){
291 mk_sched_remove_client(&l
, l
->queue
[i
].socket
);
296 /* PROCESSING CONN TIMEOUT */
297 req_idx
= mk_sched_get_request_index();
298 req_cl
= req_idx
->first
;
301 if(req_cl
->status
== MK_REQUEST_STATUS_INCOMPLETE
){
302 if((req_cl
->init_time
+ config
->timeout
) >=
304 close(req_cl
->socket
);
307 req_cl
= req_cl
->next
;