1 /*- 2 * Copyright (c) 2005 Michael Bushkov <bushman@rsu.ru> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions 7 * are met: 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. 13 * 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 24 * SUCH DAMAGE. 25 * 26 */ 27 28 #include <sys/cdefs.h> 29 __FBSDID("$FreeBSD$"); 30 31 #include <sys/socket.h> 32 #include <sys/time.h> 33 #include <sys/types.h> 34 #include <sys/event.h> 35 #include <assert.h> 36 #include <errno.h> 37 #include <stdlib.h> 38 #include <string.h> 39 #include <stdio.h> 40 41 #include "cachelib.h" 42 #include "config.h" 43 #include "debug.h" 44 #include "log.h" 45 #include "query.h" 46 #include "mp_ws_query.h" 47 #include "singletons.h" 48 49 static int on_mp_write_session_abandon_notification(struct query_state *); 50 static int on_mp_write_session_close_notification(struct query_state *); 51 static void on_mp_write_session_destroy(struct query_state *); 52 static int on_mp_write_session_mapper(struct query_state *); 53 /* int on_mp_write_session_request_read1(struct query_state *); */ 54 static int on_mp_write_session_request_read2(struct query_state *); 55 static int on_mp_write_session_request_process(struct query_state *); 56 static int on_mp_write_session_response_write1(struct query_state *); 57 static int on_mp_write_session_write_request_read1(struct query_state *); 58 static int on_mp_write_session_write_request_read2(struct query_state *); 59 static int on_mp_write_session_write_request_process(struct query_state *); 60 static int on_mp_write_session_write_response_write1(struct query_state *); 61 62 /* 63 * This function is used as the query_state's destroy_func to make the 64 * proper cleanup in case of errors. 65 */ 66 static void 67 on_mp_write_session_destroy(struct query_state *qstate) 68 { 69 70 TRACE_IN(on_mp_write_session_destroy); 71 finalize_comm_element(&qstate->request); 72 finalize_comm_element(&qstate->response); 73 74 if (qstate->mdata != NULL) { 75 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART); 76 abandon_cache_mp_write_session( 77 (cache_mp_write_session)qstate->mdata); 78 configuration_unlock_entry(qstate->config_entry, 79 CELT_MULTIPART); 80 } 81 TRACE_OUT(on_mp_write_session_destroy); 82 } 83 84 /* 85 * The functions below are used to process multipart write session initiation 86 * requests. 87 * - on_mp_write_session_request_read1 and on_mp_write_session_request_read2 88 * read the request itself 89 * - on_mp_write_session_request_process processes it 90 * - on_mp_write_session_response_write1 sends the response 91 */ 92 int 93 on_mp_write_session_request_read1(struct query_state *qstate) 94 { 95 struct cache_mp_write_session_request *c_mp_ws_request; 96 ssize_t result; 97 98 TRACE_IN(on_mp_write_session_request_read1); 99 if (qstate->kevent_watermark == 0) 100 qstate->kevent_watermark = sizeof(size_t); 101 else { 102 init_comm_element(&qstate->request, 103 CET_MP_WRITE_SESSION_REQUEST); 104 c_mp_ws_request = get_cache_mp_write_session_request( 105 &qstate->request); 106 107 result = qstate->read_func(qstate, 108 &c_mp_ws_request->entry_length, sizeof(size_t)); 109 110 if (result != sizeof(size_t)) { 111 LOG_ERR_3("on_mp_write_session_request_read1", 112 "read failed"); 113 TRACE_OUT(on_mp_write_session_request_read1); 114 return (-1); 115 } 116 117 if (BUFSIZE_INVALID(c_mp_ws_request->entry_length)) { 118 LOG_ERR_3("on_mp_write_session_request_read1", 119 "invalid entry_length value"); 120 TRACE_OUT(on_mp_write_session_request_read1); 121 return (-1); 122 } 123 124 c_mp_ws_request->entry = (char *)calloc(1, 125 c_mp_ws_request->entry_length + 1); 126 assert(c_mp_ws_request->entry != NULL); 127 128 qstate->kevent_watermark = c_mp_ws_request->entry_length; 129 qstate->process_func = on_mp_write_session_request_read2; 130 } 131 TRACE_OUT(on_mp_write_session_request_read1); 132 return (0); 133 } 134 135 static int 136 on_mp_write_session_request_read2(struct query_state *qstate) 137 { 138 struct cache_mp_write_session_request *c_mp_ws_request; 139 ssize_t result; 140 141 TRACE_IN(on_mp_write_session_request_read2); 142 c_mp_ws_request = get_cache_mp_write_session_request(&qstate->request); 143 144 result = qstate->read_func(qstate, c_mp_ws_request->entry, 145 c_mp_ws_request->entry_length); 146 147 if (result != qstate->kevent_watermark) { 148 LOG_ERR_3("on_mp_write_session_request_read2", 149 "read failed"); 150 TRACE_OUT(on_mp_write_session_request_read2); 151 return (-1); 152 } 153 154 qstate->kevent_watermark = 0; 155 qstate->process_func = on_mp_write_session_request_process; 156 157 TRACE_OUT(on_mp_write_session_request_read2); 158 return (0); 159 } 160 161 static int 162 on_mp_write_session_request_process(struct query_state *qstate) 163 { 164 struct cache_mp_write_session_request *c_mp_ws_request; 165 struct cache_mp_write_session_response *c_mp_ws_response; 166 cache_mp_write_session ws; 167 cache_entry c_entry; 168 char *dec_cache_entry_name; 169 170 TRACE_IN(on_mp_write_session_request_process); 171 init_comm_element(&qstate->response, CET_MP_WRITE_SESSION_RESPONSE); 172 c_mp_ws_response = get_cache_mp_write_session_response( 173 &qstate->response); 174 c_mp_ws_request = get_cache_mp_write_session_request(&qstate->request); 175 176 qstate->config_entry = configuration_find_entry( 177 s_configuration, c_mp_ws_request->entry); 178 if (qstate->config_entry == NULL) { 179 c_mp_ws_response->error_code = ENOENT; 180 181 LOG_ERR_2("write_session_request", 182 "can't find configuration entry '%s'. " 183 "aborting request", c_mp_ws_request->entry); 184 goto fin; 185 } 186 187 if (qstate->config_entry->enabled == 0) { 188 c_mp_ws_response->error_code = EACCES; 189 190 LOG_ERR_2("write_session_request", 191 "configuration entry '%s' is disabled", 192 c_mp_ws_request->entry); 193 goto fin; 194 } 195 196 if (qstate->config_entry->perform_actual_lookups != 0) { 197 c_mp_ws_response->error_code = EOPNOTSUPP; 198 199 LOG_ERR_2("write_session_request", 200 "entry '%s' performs lookups by itself: " 201 "can't write to it", c_mp_ws_request->entry); 202 goto fin; 203 } else { 204 #ifdef NS_NSCD_EID_CHECKING 205 if (check_query_eids(qstate) != 0) { 206 c_mp_ws_response->error_code = EPERM; 207 goto fin; 208 } 209 #endif 210 } 211 212 /* 213 * All multipart entries are separated by their name decorations. 214 * For one configuration entry there will be a lot of multipart 215 * cache entries - each with its own decorated name. 216 */ 217 asprintf(&dec_cache_entry_name, "%s%s", qstate->eid_str, 218 qstate->config_entry->mp_cache_params.entry_name); 219 assert(dec_cache_entry_name != NULL); 220 221 configuration_lock_rdlock(s_configuration); 222 c_entry = find_cache_entry(s_cache, 223 dec_cache_entry_name); 224 configuration_unlock(s_configuration); 225 226 if (c_entry == INVALID_CACHE_ENTRY) 227 c_entry = register_new_mp_cache_entry(qstate, 228 dec_cache_entry_name); 229 230 free(dec_cache_entry_name); 231 232 assert(c_entry != NULL); 233 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART); 234 ws = open_cache_mp_write_session(c_entry); 235 if (ws == INVALID_CACHE_MP_WRITE_SESSION) 236 c_mp_ws_response->error_code = -1; 237 else { 238 qstate->mdata = ws; 239 qstate->destroy_func = on_mp_write_session_destroy; 240 241 if ((qstate->config_entry->mp_query_timeout.tv_sec != 0) || 242 (qstate->config_entry->mp_query_timeout.tv_usec != 0)) 243 memcpy(&qstate->timeout, 244 &qstate->config_entry->mp_query_timeout, 245 sizeof(struct timeval)); 246 } 247 configuration_unlock_entry(qstate->config_entry, CELT_MULTIPART); 248 249 fin: 250 qstate->process_func = on_mp_write_session_response_write1; 251 qstate->kevent_watermark = sizeof(int); 252 qstate->kevent_filter = EVFILT_WRITE; 253 254 TRACE_OUT(on_mp_write_session_request_process); 255 return (0); 256 } 257 258 static int 259 on_mp_write_session_response_write1(struct query_state *qstate) 260 { 261 struct cache_mp_write_session_response *c_mp_ws_response; 262 ssize_t result; 263 264 TRACE_IN(on_mp_write_session_response_write1); 265 c_mp_ws_response = get_cache_mp_write_session_response( 266 &qstate->response); 267 result = qstate->write_func(qstate, &c_mp_ws_response->error_code, 268 sizeof(int)); 269 if (result != sizeof(int)) { 270 LOG_ERR_3("on_mp_write_session_response_write1", 271 "write failed"); 272 TRACE_OUT(on_mp_write_session_response_write1); 273 return (-1); 274 } 275 276 if (c_mp_ws_response->error_code == 0) { 277 qstate->kevent_watermark = sizeof(int); 278 qstate->process_func = on_mp_write_session_mapper; 279 qstate->kevent_filter = EVFILT_READ; 280 } else { 281 qstate->kevent_watermark = 0; 282 qstate->process_func = NULL; 283 } 284 TRACE_OUT(on_mp_write_session_response_write1); 285 return (0); 286 } 287 288 /* 289 * Mapper function is used to avoid multiple connections for each session 290 * write or read requests. After processing the request, it does not close 291 * the connection, but waits for the next request. 292 */ 293 static int 294 on_mp_write_session_mapper(struct query_state *qstate) 295 { 296 ssize_t result; 297 int elem_type; 298 299 TRACE_IN(on_mp_write_session_mapper); 300 if (qstate->kevent_watermark == 0) { 301 qstate->kevent_watermark = sizeof(int); 302 } else { 303 result = qstate->read_func(qstate, &elem_type, sizeof(int)); 304 if (result != sizeof(int)) { 305 LOG_ERR_3("on_mp_write_session_mapper", 306 "read failed"); 307 TRACE_OUT(on_mp_write_session_mapper); 308 return (-1); 309 } 310 311 switch (elem_type) { 312 case CET_MP_WRITE_SESSION_WRITE_REQUEST: 313 qstate->kevent_watermark = sizeof(size_t); 314 qstate->process_func = 315 on_mp_write_session_write_request_read1; 316 break; 317 case CET_MP_WRITE_SESSION_ABANDON_NOTIFICATION: 318 qstate->kevent_watermark = 0; 319 qstate->process_func = 320 on_mp_write_session_abandon_notification; 321 break; 322 case CET_MP_WRITE_SESSION_CLOSE_NOTIFICATION: 323 qstate->kevent_watermark = 0; 324 qstate->process_func = 325 on_mp_write_session_close_notification; 326 break; 327 default: 328 qstate->kevent_watermark = 0; 329 qstate->process_func = NULL; 330 LOG_ERR_2("on_mp_write_session_mapper", 331 "unknown element type"); 332 TRACE_OUT(on_mp_write_session_mapper); 333 return (-1); 334 } 335 } 336 TRACE_OUT(on_mp_write_session_mapper); 337 return (0); 338 } 339 340 /* 341 * The functions below are used to process multipart write sessions write 342 * requests. 343 * - on_mp_write_session_write_request_read1 and 344 * on_mp_write_session_write_request_read2 read the request itself 345 * - on_mp_write_session_write_request_process processes it 346 * - on_mp_write_session_write_response_write1 sends the response 347 */ 348 static int 349 on_mp_write_session_write_request_read1(struct query_state *qstate) 350 { 351 struct cache_mp_write_session_write_request *write_request; 352 ssize_t result; 353 354 TRACE_IN(on_mp_write_session_write_request_read1); 355 init_comm_element(&qstate->request, 356 CET_MP_WRITE_SESSION_WRITE_REQUEST); 357 write_request = get_cache_mp_write_session_write_request( 358 &qstate->request); 359 360 result = qstate->read_func(qstate, &write_request->data_size, 361 sizeof(size_t)); 362 363 if (result != sizeof(size_t)) { 364 LOG_ERR_3("on_mp_write_session_write_request_read1", 365 "read failed"); 366 TRACE_OUT(on_mp_write_session_write_request_read1); 367 return (-1); 368 } 369 370 if (BUFSIZE_INVALID(write_request->data_size)) { 371 LOG_ERR_3("on_mp_write_session_write_request_read1", 372 "invalid data_size value"); 373 TRACE_OUT(on_mp_write_session_write_request_read1); 374 return (-1); 375 } 376 377 write_request->data = (char *)calloc(1, write_request->data_size); 378 assert(write_request->data != NULL); 379 380 qstate->kevent_watermark = write_request->data_size; 381 qstate->process_func = on_mp_write_session_write_request_read2; 382 TRACE_OUT(on_mp_write_session_write_request_read1); 383 return (0); 384 } 385 386 static int 387 on_mp_write_session_write_request_read2(struct query_state *qstate) 388 { 389 struct cache_mp_write_session_write_request *write_request; 390 ssize_t result; 391 392 TRACE_IN(on_mp_write_session_write_request_read2); 393 write_request = get_cache_mp_write_session_write_request( 394 &qstate->request); 395 396 result = qstate->read_func(qstate, write_request->data, 397 write_request->data_size); 398 399 if (result != qstate->kevent_watermark) { 400 LOG_ERR_3("on_mp_write_session_write_request_read2", 401 "read failed"); 402 TRACE_OUT(on_mp_write_session_write_request_read2); 403 return (-1); 404 } 405 406 qstate->kevent_watermark = 0; 407 qstate->process_func = on_mp_write_session_write_request_process; 408 TRACE_OUT(on_mp_write_session_write_request_read2); 409 return (0); 410 } 411 412 static int 413 on_mp_write_session_write_request_process(struct query_state *qstate) 414 { 415 struct cache_mp_write_session_write_request *write_request; 416 struct cache_mp_write_session_write_response *write_response; 417 418 TRACE_IN(on_mp_write_session_write_request_process); 419 init_comm_element(&qstate->response, 420 CET_MP_WRITE_SESSION_WRITE_RESPONSE); 421 write_response = get_cache_mp_write_session_write_response( 422 &qstate->response); 423 write_request = get_cache_mp_write_session_write_request( 424 &qstate->request); 425 426 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART); 427 write_response->error_code = cache_mp_write( 428 (cache_mp_write_session)qstate->mdata, 429 write_request->data, 430 write_request->data_size); 431 configuration_unlock_entry(qstate->config_entry, CELT_MULTIPART); 432 433 qstate->kevent_watermark = sizeof(int); 434 qstate->process_func = on_mp_write_session_write_response_write1; 435 qstate->kevent_filter = EVFILT_WRITE; 436 437 TRACE_OUT(on_mp_write_session_write_request_process); 438 return (0); 439 } 440 441 static int 442 on_mp_write_session_write_response_write1(struct query_state *qstate) 443 { 444 struct cache_mp_write_session_write_response *write_response; 445 ssize_t result; 446 447 TRACE_IN(on_mp_write_session_write_response_write1); 448 write_response = get_cache_mp_write_session_write_response( 449 &qstate->response); 450 result = qstate->write_func(qstate, &write_response->error_code, 451 sizeof(int)); 452 if (result != sizeof(int)) { 453 LOG_ERR_3("on_mp_write_session_write_response_write1", 454 "write failed"); 455 TRACE_OUT(on_mp_write_session_write_response_write1); 456 return (-1); 457 } 458 459 if (write_response->error_code == 0) { 460 finalize_comm_element(&qstate->request); 461 finalize_comm_element(&qstate->response); 462 463 qstate->kevent_watermark = sizeof(int); 464 qstate->process_func = on_mp_write_session_mapper; 465 qstate->kevent_filter = EVFILT_READ; 466 } else { 467 qstate->kevent_watermark = 0; 468 qstate->process_func = 0; 469 } 470 471 TRACE_OUT(on_mp_write_session_write_response_write1); 472 return (0); 473 } 474 475 /* 476 * Handles abandon notifications. Destroys the session by calling the 477 * abandon_cache_mp_write_session. 478 */ 479 static int 480 on_mp_write_session_abandon_notification(struct query_state *qstate) 481 { 482 TRACE_IN(on_mp_write_session_abandon_notification); 483 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART); 484 abandon_cache_mp_write_session((cache_mp_write_session)qstate->mdata); 485 configuration_unlock_entry(qstate->config_entry, CELT_MULTIPART); 486 qstate->mdata = INVALID_CACHE_MP_WRITE_SESSION; 487 488 qstate->kevent_watermark = 0; 489 qstate->process_func = NULL; 490 TRACE_OUT(on_mp_write_session_abandon_notification); 491 return (0); 492 } 493 494 /* 495 * Handles close notifications. Commits the session by calling 496 * the close_cache_mp_write_session. 497 */ 498 static int 499 on_mp_write_session_close_notification(struct query_state *qstate) 500 { 501 TRACE_IN(on_mp_write_session_close_notification); 502 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART); 503 close_cache_mp_write_session((cache_mp_write_session)qstate->mdata); 504 configuration_unlock_entry(qstate->config_entry, CELT_MULTIPART); 505 qstate->mdata = INVALID_CACHE_MP_WRITE_SESSION; 506 507 qstate->kevent_watermark = 0; 508 qstate->process_func = NULL; 509 TRACE_OUT(on_mp_write_session_close_notification); 510 return (0); 511 } 512 513 cache_entry register_new_mp_cache_entry(struct query_state *qstate, 514 const char *dec_cache_entry_name) 515 { 516 cache_entry c_entry; 517 char *en_bkp; 518 519 TRACE_IN(register_new_mp_cache_entry); 520 c_entry = INVALID_CACHE_ENTRY; 521 configuration_lock_entry(qstate->config_entry, CELT_MULTIPART); 522 523 configuration_lock_wrlock(s_configuration); 524 en_bkp = qstate->config_entry->mp_cache_params.entry_name; 525 qstate->config_entry->mp_cache_params.entry_name = 526 (char *)dec_cache_entry_name; 527 register_cache_entry(s_cache, (struct cache_entry_params *) 528 &qstate->config_entry->mp_cache_params); 529 qstate->config_entry->mp_cache_params.entry_name = en_bkp; 530 configuration_unlock(s_configuration); 531 532 configuration_lock_rdlock(s_configuration); 533 c_entry = find_cache_entry(s_cache, 534 dec_cache_entry_name); 535 configuration_unlock(s_configuration); 536 537 configuration_entry_add_mp_cache_entry(qstate->config_entry, 538 c_entry); 539 540 configuration_unlock_entry(qstate->config_entry, 541 CELT_MULTIPART); 542 543 TRACE_OUT(register_new_mp_cache_entry); 544 return (c_entry); 545 } 546