2 * Copyright (c) 2005 Michael Bushkov <bushman@rsu.ru>
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
8 * 1. Redistributions of source code must retain the above copyright
9 * notice, this list of conditions and the following disclaimer.
10 * 2. Redistributions in binary form must reproduce the above copyright
11 * notice, this list of conditions and the following disclaimer in the
12 * documentation and/or other materials provided with the distribution.
14 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
15 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
16 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
17 * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
18 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
19 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
20 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
21 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
22 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
23 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
28 #include <sys/cdefs.h>
29 __FBSDID("$FreeBSD$");
31 #include <sys/types.h>
32 #include <sys/event.h>
33 #include <sys/socket.h>
48 #include "mp_rs_query.h"
49 #include "mp_ws_query.h"
50 #include "singletons.h"
52 static int on_mp_read_session_close_notification(struct query_state *);
53 static void on_mp_read_session_destroy(struct query_state *);
54 static int on_mp_read_session_mapper(struct query_state *);
55 /* int on_mp_read_session_request_read1(struct query_state *); */
56 static int on_mp_read_session_request_read2(struct query_state *);
57 static int on_mp_read_session_request_process(struct query_state *);
58 static int on_mp_read_session_response_write1(struct query_state *);
59 static int on_mp_read_session_read_request_process(struct query_state *);
60 static int on_mp_read_session_read_response_write1(struct query_state *);
61 static int on_mp_read_session_read_response_write2(struct query_state *);
64 * This function is used as the query_state's destroy_func to make the
65 * proper cleanup in case of errors.
68 on_mp_read_session_destroy(struct query_state *qstate)
70 TRACE_IN(on_mp_read_session_destroy);
71 finalize_comm_element(&qstate->request);
72 finalize_comm_element(&qstate->response);
74 if (qstate->mdata != NULL) {
75 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART);
76 close_cache_mp_read_session(
77 (cache_mp_read_session)qstate->mdata);
78 configuration_unlock_entry(qstate->config_entry,
81 TRACE_OUT(on_mp_read_session_destroy);
85 * The functions below are used to process multipart read session initiation
87 * - on_mp_read_session_request_read1 and on_mp_read_session_request_read2 read
89 * - on_mp_read_session_request_process processes it
90 * - on_mp_read_session_response_write1 sends the response
93 on_mp_read_session_request_read1(struct query_state *qstate)
95 struct cache_mp_read_session_request *c_mp_rs_request;
98 TRACE_IN(on_mp_read_session_request_read1);
99 if (qstate->kevent_watermark == 0)
100 qstate->kevent_watermark = sizeof(size_t);
102 init_comm_element(&qstate->request,
103 CET_MP_READ_SESSION_REQUEST);
104 c_mp_rs_request = get_cache_mp_read_session_request(
107 result = qstate->read_func(qstate,
108 &c_mp_rs_request->entry_length, sizeof(size_t));
110 if (result != sizeof(size_t)) {
111 TRACE_OUT(on_mp_read_session_request_read1);
115 if (BUFSIZE_INVALID(c_mp_rs_request->entry_length)) {
116 TRACE_OUT(on_mp_read_session_request_read1);
120 c_mp_rs_request->entry = calloc(1,
121 c_mp_rs_request->entry_length + 1);
122 assert(c_mp_rs_request->entry != NULL);
124 qstate->kevent_watermark = c_mp_rs_request->entry_length;
125 qstate->process_func = on_mp_read_session_request_read2;
127 TRACE_OUT(on_mp_read_session_request_read1);
132 on_mp_read_session_request_read2(struct query_state *qstate)
134 struct cache_mp_read_session_request *c_mp_rs_request;
137 TRACE_IN(on_mp_read_session_request_read2);
138 c_mp_rs_request = get_cache_mp_read_session_request(&qstate->request);
140 result = qstate->read_func(qstate, c_mp_rs_request->entry,
141 c_mp_rs_request->entry_length);
143 if (result < 0 || (size_t)result != qstate->kevent_watermark) {
144 LOG_ERR_3("on_mp_read_session_request_read2",
146 TRACE_OUT(on_mp_read_session_request_read2);
150 qstate->kevent_watermark = 0;
151 qstate->process_func = on_mp_read_session_request_process;
152 TRACE_OUT(on_mp_read_session_request_read2);
157 on_mp_read_session_request_process(struct query_state *qstate)
159 struct cache_mp_read_session_request *c_mp_rs_request;
160 struct cache_mp_read_session_response *c_mp_rs_response;
161 cache_mp_read_session rs;
163 char *dec_cache_entry_name;
167 cache_mp_write_session ws;
168 struct agent *lookup_agent;
169 struct multipart_agent *mp_agent;
173 TRACE_IN(on_mp_read_session_request_process);
174 init_comm_element(&qstate->response, CET_MP_READ_SESSION_RESPONSE);
175 c_mp_rs_response = get_cache_mp_read_session_response(
177 c_mp_rs_request = get_cache_mp_read_session_request(&qstate->request);
179 qstate->config_entry = configuration_find_entry(
180 s_configuration, c_mp_rs_request->entry);
181 if (qstate->config_entry == NULL) {
182 c_mp_rs_response->error_code = ENOENT;
184 LOG_ERR_2("read_session_request",
185 "can't find configuration entry '%s'."
186 " aborting request", c_mp_rs_request->entry);
190 if (qstate->config_entry->enabled == 0) {
191 c_mp_rs_response->error_code = EACCES;
193 LOG_ERR_2("read_session_request",
194 "configuration entry '%s' is disabled",
195 c_mp_rs_request->entry);
199 if (qstate->config_entry->perform_actual_lookups != 0)
200 dec_cache_entry_name = strdup(
201 qstate->config_entry->mp_cache_params.cep.entry_name);
203 #ifdef NS_NSCD_EID_CHECKING
204 if (check_query_eids(qstate) != 0) {
205 c_mp_rs_response->error_code = EPERM;
210 asprintf(&dec_cache_entry_name, "%s%s", qstate->eid_str,
211 qstate->config_entry->mp_cache_params.cep.entry_name);
214 assert(dec_cache_entry_name != NULL);
216 configuration_lock_rdlock(s_configuration);
217 c_entry = find_cache_entry(s_cache, dec_cache_entry_name);
218 configuration_unlock(s_configuration);
220 if ((c_entry == INVALID_CACHE) &&
221 (qstate->config_entry->perform_actual_lookups != 0))
222 c_entry = register_new_mp_cache_entry(qstate,
223 dec_cache_entry_name);
225 free(dec_cache_entry_name);
227 if (c_entry != INVALID_CACHE_ENTRY) {
228 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART);
229 rs = open_cache_mp_read_session(c_entry);
230 configuration_unlock_entry(qstate->config_entry,
233 if ((rs == INVALID_CACHE_MP_READ_SESSION) &&
234 (qstate->config_entry->perform_actual_lookups != 0)) {
235 lookup_agent = find_agent(s_agent_table,
236 c_mp_rs_request->entry, MULTIPART_AGENT);
238 if ((lookup_agent != NULL) &&
239 (lookup_agent->type == MULTIPART_AGENT)) {
240 mp_agent = (struct multipart_agent *)
242 mdata = mp_agent->mp_init_func();
245 * Multipart agents read the whole snapshot
246 * of the data at one time.
248 configuration_lock_entry(qstate->config_entry,
250 ws = open_cache_mp_write_session(c_entry);
251 configuration_unlock_entry(qstate->config_entry,
256 res = mp_agent->mp_lookup_func(&buffer,
260 if ((res & NS_TERMINATE) &&
262 configuration_lock_entry(
263 qstate->config_entry,
265 if (cache_mp_write(ws, buffer,
267 abandon_cache_mp_write_session(ws);
270 configuration_unlock_entry(
271 qstate->config_entry,
277 configuration_lock_entry(
278 qstate->config_entry,
280 close_cache_mp_write_session(ws);
281 configuration_unlock_entry(
282 qstate->config_entry,
288 } while ((res & NS_TERMINATE) &&
292 configuration_lock_entry(qstate->config_entry,
294 rs = open_cache_mp_read_session(c_entry);
295 configuration_unlock_entry(qstate->config_entry,
300 if (rs == INVALID_CACHE_MP_READ_SESSION)
301 c_mp_rs_response->error_code = -1;
304 qstate->destroy_func = on_mp_read_session_destroy;
306 configuration_lock_entry(qstate->config_entry,
308 if ((qstate->config_entry->mp_query_timeout.tv_sec != 0) ||
309 (qstate->config_entry->mp_query_timeout.tv_usec != 0))
310 memcpy(&qstate->timeout,
311 &qstate->config_entry->mp_query_timeout,
312 sizeof(struct timeval));
313 configuration_unlock_entry(qstate->config_entry,
317 c_mp_rs_response->error_code = -1;
320 qstate->process_func = on_mp_read_session_response_write1;
321 qstate->kevent_watermark = sizeof(int);
322 qstate->kevent_filter = EVFILT_WRITE;
324 TRACE_OUT(on_mp_read_session_request_process);
329 on_mp_read_session_response_write1(struct query_state *qstate)
331 struct cache_mp_read_session_response *c_mp_rs_response;
334 TRACE_IN(on_mp_read_session_response_write1);
335 c_mp_rs_response = get_cache_mp_read_session_response(
337 result = qstate->write_func(qstate, &c_mp_rs_response->error_code,
340 if (result != sizeof(int)) {
341 LOG_ERR_3("on_mp_read_session_response_write1",
343 TRACE_OUT(on_mp_read_session_response_write1);
347 if (c_mp_rs_response->error_code == 0) {
348 qstate->kevent_watermark = sizeof(int);
349 qstate->process_func = on_mp_read_session_mapper;
350 qstate->kevent_filter = EVFILT_READ;
352 qstate->kevent_watermark = 0;
353 qstate->process_func = NULL;
355 TRACE_OUT(on_mp_read_session_response_write1);
360 * Mapper function is used to avoid multiple connections for each session
361 * write or read requests. After processing the request, it does not close
362 * the connection, but waits for the next request.
365 on_mp_read_session_mapper(struct query_state *qstate)
370 TRACE_IN(on_mp_read_session_mapper);
371 if (qstate->kevent_watermark == 0) {
372 qstate->kevent_watermark = sizeof(int);
374 result = qstate->read_func(qstate, &elem_type, sizeof(int));
375 if (result != sizeof(int)) {
376 LOG_ERR_3("on_mp_read_session_mapper",
378 TRACE_OUT(on_mp_read_session_mapper);
383 case CET_MP_READ_SESSION_READ_REQUEST:
384 qstate->kevent_watermark = 0;
385 qstate->process_func =
386 on_mp_read_session_read_request_process;
388 case CET_MP_READ_SESSION_CLOSE_NOTIFICATION:
389 qstate->kevent_watermark = 0;
390 qstate->process_func =
391 on_mp_read_session_close_notification;
394 qstate->kevent_watermark = 0;
395 qstate->process_func = NULL;
396 LOG_ERR_3("on_mp_read_session_mapper",
397 "unknown element type");
398 TRACE_OUT(on_mp_read_session_mapper);
402 TRACE_OUT(on_mp_read_session_mapper);
407 * The functions below are used to process multipart read sessions read
408 * requests. User doesn't have to pass any kind of data, besides the
409 * request identificator itself. So we don't need any XXX_read functions and
410 * start with the XXX_process function.
411 * - on_mp_read_session_read_request_process processes it
412 * - on_mp_read_session_read_response_write1 and
413 * on_mp_read_session_read_response_write2 sends the response
416 on_mp_read_session_read_request_process(struct query_state *qstate)
418 struct cache_mp_read_session_read_response *read_response;
420 TRACE_IN(on_mp_read_session_response_process);
421 init_comm_element(&qstate->response, CET_MP_READ_SESSION_READ_RESPONSE);
422 read_response = get_cache_mp_read_session_read_response(
425 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART);
426 read_response->error_code = cache_mp_read(
427 (cache_mp_read_session)qstate->mdata, NULL,
428 &read_response->data_size);
430 if (read_response->error_code == 0) {
431 read_response->data = malloc(read_response->data_size);
432 assert(read_response != NULL);
433 read_response->error_code = cache_mp_read(
434 (cache_mp_read_session)qstate->mdata,
436 &read_response->data_size);
438 configuration_unlock_entry(qstate->config_entry, CELT_MULTIPART);
440 if (read_response->error_code == 0)
441 qstate->kevent_watermark = sizeof(size_t) + sizeof(int);
443 qstate->kevent_watermark = sizeof(int);
444 qstate->process_func = on_mp_read_session_read_response_write1;
445 qstate->kevent_filter = EVFILT_WRITE;
447 TRACE_OUT(on_mp_read_session_response_process);
452 on_mp_read_session_read_response_write1(struct query_state *qstate)
454 struct cache_mp_read_session_read_response *read_response;
457 TRACE_IN(on_mp_read_session_read_response_write1);
458 read_response = get_cache_mp_read_session_read_response(
461 result = qstate->write_func(qstate, &read_response->error_code,
463 if (read_response->error_code == 0) {
464 result += qstate->write_func(qstate, &read_response->data_size,
466 if (result < 0 || (size_t)result != qstate->kevent_watermark) {
467 TRACE_OUT(on_mp_read_session_read_response_write1);
468 LOG_ERR_3("on_mp_read_session_read_response_write1",
473 qstate->kevent_watermark = read_response->data_size;
474 qstate->process_func = on_mp_read_session_read_response_write2;
476 if (result < 0 || (size_t)result != qstate->kevent_watermark) {
477 LOG_ERR_3("on_mp_read_session_read_response_write1",
479 TRACE_OUT(on_mp_read_session_read_response_write1);
483 qstate->kevent_watermark = 0;
484 qstate->process_func = NULL;
487 TRACE_OUT(on_mp_read_session_read_response_write1);
492 on_mp_read_session_read_response_write2(struct query_state *qstate)
494 struct cache_mp_read_session_read_response *read_response;
497 TRACE_IN(on_mp_read_session_read_response_write2);
498 read_response = get_cache_mp_read_session_read_response(
500 result = qstate->write_func(qstate, read_response->data,
501 read_response->data_size);
502 if (result < 0 || (size_t)result != qstate->kevent_watermark) {
503 LOG_ERR_3("on_mp_read_session_read_response_write2",
505 TRACE_OUT(on_mp_read_session_read_response_write2);
509 finalize_comm_element(&qstate->request);
510 finalize_comm_element(&qstate->response);
512 qstate->kevent_watermark = sizeof(int);
513 qstate->process_func = on_mp_read_session_mapper;
514 qstate->kevent_filter = EVFILT_READ;
516 TRACE_OUT(on_mp_read_session_read_response_write2);
521 * Handles session close notification by calling close_cache_mp_read_session
525 on_mp_read_session_close_notification(struct query_state *qstate)
528 TRACE_IN(on_mp_read_session_close_notification);
529 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART);
530 close_cache_mp_read_session((cache_mp_read_session)qstate->mdata);
531 configuration_unlock_entry(qstate->config_entry, CELT_MULTIPART);
532 qstate->mdata = NULL;
533 qstate->kevent_watermark = 0;
534 qstate->process_func = NULL;
535 TRACE_OUT(on_mp_read_session_close_notification);